diff --git a/benchmarks/base.py b/benchmarks/base.py
index 77a168b0..66ca724d 100644
--- a/benchmarks/base.py
+++ b/benchmarks/base.py
@@ -124,7 +124,9 @@ def benchmark(thread_class):
         start = time.time()
         try:
             for i in range(options.threads):
-                thread = thread_class(i, session, query, values, per_thread, options.profile)
+                thread = thread_class(
+                    i, session, query, values, per_thread,
+                    cluster.protocol_version, options.profile)
                 thread.daemon = True
                 threads.append(thread)
 
@@ -217,13 +219,14 @@ def parse_options():
 
 class BenchmarkThread(Thread):
 
-    def __init__(self, thread_num, session, query, values, num_queries, profile):
+    def __init__(self, thread_num, session, query, values, num_queries, protocol_version, profile):
         Thread.__init__(self)
         self.thread_num = thread_num
         self.session = session
         self.query = query
         self.values = values
         self.num_queries = num_queries
+        self.protocol_version = protocol_version
         self.profiler = Profile() if profile else None
 
     def start_profile(self):
diff --git a/benchmarks/callback_full_pipeline.py b/benchmarks/callback_full_pipeline.py
index bf84077b..c79a6a34 100644
--- a/benchmarks/callback_full_pipeline.py
+++ b/benchmarks/callback_full_pipeline.py
@@ -48,7 +48,12 @@ class Runner(BenchmarkThread):
     def run(self):
         self.start_profile()
 
-        for _ in range(min(120, self.num_queries)):
+        if self.protocol_version >= 3:
+            concurrency = 1000
+        else:
+            concurrency = 100
+
+        for _ in range(min(concurrency, self.num_queries)):
             self.insert_next()
 
         self.event.wait()