Add grn_array_unblock()

It unblocks rb_array_pull().

include/groonga.h

@@ -2844,6 +2844,7 @@ GRN_API grn_id grn_array_pull(grn_ctx *ctx, grn_array *array, grn_bool blockp,
                               void (*func)(grn_ctx *ctx, grn_array *array,
                                            grn_id id, void *func_arg),
                               void *func_arg);
+GRN_API void grn_array_unblock(grn_ctx *ctx, grn_array *array);
 GRN_API int grn_array_get_value(grn_ctx *ctx, grn_array *array, grn_id id, void *valuebuf);
 GRN_API grn_rc grn_array_set_value(grn_ctx *ctx, grn_array *array, grn_id id,
                                    const void *value, int flags);

lib/hash.c

@@ -336,6 +336,7 @@ grn_table_queue_init(grn_ctx *ctx, grn_table_queue *queue)
   queue->head = 0;
   queue->tail = 0;
   queue->cap = GRN_ARRAY_MAX;
+  queue->unblock_requested = GRN_FALSE;
   grn_table_queue_lock_init(ctx, queue);
 }
 
@@ -1130,8 +1131,9 @@ grn_array_pull(grn_ctx *ctx, grn_array *array, grn_bool blockp,
   grn_table_queue *queue = grn_array_queue(ctx, array);
   if (queue) {
     MUTEX_LOCK(queue->mutex);
+    queue->unblock_requested = GRN_FALSE;
     while (grn_table_queue_size(queue) == 0) {
-      if (!blockp) {
+      if (!blockp || queue->unblock_requested) {
         MUTEX_UNLOCK(queue->mutex);
         GRN_OUTPUT_BOOL(0);
         return id;
@@ -1150,6 +1152,18 @@ grn_array_pull(grn_ctx *ctx, grn_array *array, grn_bool blockp,
   return id;
 }
 
+void
+grn_array_unblock(grn_ctx *ctx, grn_array *array)
+{
+  grn_table_queue *queue = grn_array_queue(ctx, array);
+  if (!queue) {
+    return;
+  }
+
+  queue->unblock_requested = GRN_TRUE;
+  COND_BROADCAST(queue->cond);
+}
+
 /* grn_hash : hash table */
 
 #define GRN_HASH_MAX_SEGMENT  0x400

lib/hash.h

@@ -177,6 +177,7 @@ struct _grn_table_queue {
   grn_id head;
   grn_id tail;
   grn_id cap;
+  grn_bool unblock_requested;
 };
 
 GRN_API void grn_array_queue_lock_clear(grn_ctx *ctx, grn_array *array);