Dedicated queue writers

Increasing the number of writers on the master does not always improve
database performance. More often than not it increases contention
and hurts performance. But queues may be an exception: only one client
can consume from a queue at a time, that it could make sense for
queue consumers to run in parellel as much as possible.

On a tunable, this patch dispatches queue consumes to dedicated
writer threads. The handle_buf code is also refactored to use thdpool.

Signed-off-by: Rivers Zhang <[email protected]>

Author: riverszhang89

Diff for: bbinc/thdpool.h

@@ -68,7 +68,8 @@ typedef void (*thdpool_thddque_fn)(struct thdpool *pool, struct workitem *item,
 typedef void (*thdpool_foreach_fn)(struct thdpool *pool, struct workitem *item,
                                    void *user);
 void thdpool_foreach(struct thdpool *pool, thdpool_foreach_fn, void *user);
-
+typedef void (*thdpool_for_each_thd_fn)(struct thdpool *pool, pthread_t tid, int idle, void *thddata, void *user);
+void thdpool_for_each_thd(struct thdpool *pool, thdpool_for_each_thd_fn, void *user);
 struct thdpool *thdpool_create(const char *name, size_t per_thread_data_sz);
 int thdpool_destroy(struct thdpool **pool_p, int coopWaitUs);
 void thdpool_set_stack_size(struct thdpool *pool, size_t sz_bytes);

Diff for: bdb/bdb_thd_io.c

@@ -156,29 +156,6 @@ static ssize_t bdb_write(int fd, const void *buf, size_t nbytes)
     return rc;
 }
 
-void bdb_set_io_control(void (*start)(), void (*cmplt)())
-{
-    logmsg(LOGMSG_DEBUG, "IO CONTROL disabled\n");
-    if (1)
-        return;
-#if 0   
-   io_start=start;
-   io_cmplt=cmplt;
-   
-   if 
-     (
-      db_env_set_func_read(bdb_read) ||
-      db_env_set_func_fsync(bdb_fsync) ||
-      db_env_set_func_write(bdb_write)
-      )
-   {
-      printf("**FAILED db_env_set_func_pread RC\n");
-      abort();
-   }
-   printf("SET UP THD IO CONTROL\n");
-#endif
-}
-
 void bdb_get_iostats(int *n_reads, int *l_reads, int *n_writes, int *l_writes)
 {
     *n_reads = norm_reads;

Diff for: db/comdb2.h

@@ -1287,6 +1287,7 @@ struct osql_sess {
     pthread_mutex_t participant_lk;
     int is_done;
     int is_sanctioned; /* set by fdb from coordinator-master */
+    int is_qconsume_only;
 };
 typedef struct osql_sess osql_sess_t;
 
@@ -1823,6 +1824,9 @@ extern int gbl_appsock_pooling;
 extern struct thdpool *gbl_appsock_thdpool;
 extern struct thdpool *gbl_osqlpfault_thdpool;
 extern struct thdpool *gbl_udppfault_thdpool;
+extern struct thdpool *gbl_handle_buf_write_thdpool;
+extern struct thdpool *gbl_handle_buf_read_thdpool;
+extern struct thdpool *gbl_handle_buf_queue_thdpool;
 
 extern int gbl_consumer_rtcpu_check;
 extern int gbl_node1rtcpuable;

Diff for: db/db_tunables.c

@@ -564,6 +564,8 @@ extern int gbl_sc_history_max_rows;
 extern int gbl_sc_status_max_rows;
 extern int gbl_rep_process_pstack_time;
 extern int gbl_sql_recover_time;
+extern int gbl_queue_use_dedicated_writers;
+extern int gbl_queue_max_dedicated_writers;
 
 extern void set_snapshot_impl(snap_impl_enum impl);
 extern const char *snap_impl_str(snap_impl_enum impl);

Diff for: db/db_tunables.h

@@ -2485,4 +2485,6 @@ REGISTER_TUNABLE("sc_status_max_rows", "Max number of rows returned in comdb2_sc
 REGISTER_TUNABLE("rep_process_pstack_time", "pstack the server if rep_process runs longer than time specified in secs (Default: 30s)",
                  TUNABLE_INTEGER, &gbl_rep_process_pstack_time, 0, NULL, NULL, NULL, NULL);
 REGISTER_TUNABLE("sql_recover_time", "Number of msec before checking if SQL has waiters. 0 will disable. (Default: 10ms)", TUNABLE_INTEGER, &gbl_sql_recover_time, 0, NULL, NULL, NULL, NULL);
+REGISTER_TUNABLE("queue_use_dedicated_writers", "Whether queue-consumes are processed in dedicated writers. (Default: on)", TUNABLE_BOOLEAN, &gbl_queue_use_dedicated_writers, 0, NULL, NULL, NULL, NULL);
+REGISTER_TUNABLE("queue_max_dedicated_writers", "Max number of dedicated queue-consume writers. (Default: on)", TUNABLE_INTEGER, &gbl_queue_max_dedicated_writers, 0, NULL, NULL, NULL, NULL);
 #endif /* _DB_TUNABLES_H */