add files: common_blocked_queue.[hc]

HISTORY

@@ -1,8 +1,9 @@
 
-Version 1.38  2018-05-11
+Version 1.38  2018-05-17
   * connection_pool.c: set err_no to 0 when success
   * shared_func.h: add functions float2buff / buff2float, double2buff / buff2double
   * logger.h: add function log_get_level_caption
+  * add files: common_blocked_queue.[hc]
 
 Version 1.37  2018-02-24
   * ini_file_reader.c function annotations LOCAL_IP_GET support index, such as:

src/Makefile.in

@@ -12,7 +12,7 @@ FAST_SHARED_OBJS = hash.lo chain.lo shared_func.lo ini_file_reader.lo \
                    connection_pool.lo fast_mpool.lo fast_allocator.lo  \
                    fast_buffer.lo multi_skiplist.lo flat_skiplist.lo \
                    system_info.lo fast_blocked_queue.lo id_generator.lo \
-				   char_converter.lo char_convert_loader.lo
+				   char_converter.lo char_convert_loader.lo common_blocked_queue.lo
 
 FAST_STATIC_OBJS = hash.o chain.o shared_func.o ini_file_reader.o \
                    logger.o sockopt.o base64.o sched_thread.o \
@@ -22,7 +22,7 @@ FAST_STATIC_OBJS = hash.o chain.o shared_func.o ini_file_reader.o \
                    connection_pool.o fast_mpool.o fast_allocator.o \
                    fast_buffer.o multi_skiplist.o flat_skiplist.o  \
                    system_info.o fast_blocked_queue.o id_generator.o \
-				   char_converter.o char_convert_loader.o
+				   char_converter.o char_convert_loader.o common_blocked_queue.o
 
 HEADER_FILES = common_define.h hash.h chain.h logger.h base64.h \
                shared_func.h pthread_func.h ini_file_reader.h _os_define.h \
@@ -33,7 +33,7 @@ HEADER_FILES = common_define.h hash.h chain.h logger.h base64.h \
                fast_buffer.h skiplist.h multi_skiplist.h flat_skiplist.h \
                skiplist_common.h system_info.h fast_blocked_queue.h \
                php7_ext_wrapper.h id_generator.h char_converter.h \
-			   char_convert_loader.h
+			   char_convert_loader.h common_blocked_queue.h
 
 ALL_OBJS = $(FAST_STATIC_OBJS) $(FAST_SHARED_OBJS)
 
@@ -59,9 +59,8 @@ install:
 	mkdir -p $(DESTDIR)/usr/$(LIB_VERSION)
 	mkdir -p $(DESTDIR)/usr/lib
 	install -m 755 $(SHARED_LIBS) $(DESTDIR)/usr/$(LIB_VERSION)
-	install -m 755 $(SHARED_LIBS) $(DESTDIR)/usr/lib
+	mkdir -p $(DESTDIR)/usr/local/include/fastcommon
-	mkdir -p $(DESTDIR)/usr/include/fastcommon
+	install -m 644 $(HEADER_FILES) $(DESTDIR)/usr/local/include/fastcommon
-	install -m 644 $(HEADER_FILES) $(DESTDIR)/usr/include/fastcommon
 clean:
 	rm -f $(ALL_OBJS) $(ALL_PRGS) $(ALL_LIBS)
 

src/common_blocked_queue.c

@@ -0,0 +1,153 @@
+//common_blocked_queue.c
+
+#include <errno.h>
+#include <pthread.h>
+#include <inttypes.h>
+#include "logger.h"
+#include "shared_func.h"
+#include "pthread_func.h"
+#include "common_blocked_queue.h"
+
+int common_blocked_queue_init_ex(struct common_blocked_queue *queue,
+        const int alloc_elements_once)
+{
+	int result;
+
+	if ((result=init_pthread_lock(&(queue->lock))) != 0)
+	{
+		logError("file: "__FILE__", line: %d, "
+			"init_pthread_lock fail, errno: %d, error info: %s",
+			__LINE__, result, STRERROR(result));
+		return result;
+	}
+
+    result = pthread_cond_init(&(queue->cond), NULL);
+    if (result != 0)
+    {
+        logError("file: "__FILE__", line: %d, "
+                "pthread_cond_init fail, "
+                "errno: %d, error info: %s",
+                __LINE__, result, STRERROR(result));
+        return result;
+    }
+
+    if ((result=fast_mblock_init_ex(&queue->mblock,
+                    sizeof(struct common_blocked_node),
+                    alloc_elements_once, NULL, false)) != 0)
+    {
+        return result;
+    }
+
+	queue->head = NULL;
+	queue->tail = NULL;
+
+	return 0;
+}
+
+void common_blocked_queue_destroy(struct common_blocked_queue *queue)
+{
+    pthread_cond_destroy(&(queue->cond));
+    pthread_mutex_destroy(&(queue->lock));
+}
+
+int common_blocked_queue_push(struct common_blocked_queue *queue, void *data)
+{
+	int result;
+    struct common_blocked_node *node;
+    bool notify;
+
+	if ((result=pthread_mutex_lock(&(queue->lock))) != 0)
+	{
+		logError("file: "__FILE__", line: %d, " \
+			"call pthread_mutex_lock fail, " \
+			"errno: %d, error info: %s", \
+			__LINE__, result, STRERROR(result));
+		return result;
+	}
+
+    node = (struct common_blocked_node *)fast_mblock_alloc_object(&queue->mblock);
+    if (node == NULL)
+    {
+        pthread_mutex_unlock(&(queue->lock));
+		return ENOMEM;
+    }
+
+	node->data = data;
+	node->next = NULL;
+
+	if (queue->tail == NULL)
+	{
+		queue->head = node;
+        notify = true;
+	}
+	else
+	{
+		queue->tail->next = node;
+        notify = false;
+	}
+	queue->tail = node;
+
+	if ((result=pthread_mutex_unlock(&(queue->lock))) != 0)
+	{
+		logError("file: "__FILE__", line: %d, " \
+			"call pthread_mutex_unlock fail, " \
+			"errno: %d, error info: %s", \
+			__LINE__, result, STRERROR(result));
+	}
+
+    if (notify)
+    {
+        pthread_cond_signal(&(queue->cond));
+    }
+
+	return 0;
+}
+
+void *common_blocked_queue_pop(struct common_blocked_queue *queue)
+{
+    struct common_blocked_node *node;
+	void *data;
+	int result;
+
+	if ((result=pthread_mutex_lock(&(queue->lock))) != 0)
+	{
+		logError("file: "__FILE__", line: %d, " \
+			"call pthread_mutex_lock fail, " \
+			"errno: %d, error info: %s", \
+			__LINE__, result, STRERROR(result));
+		return NULL;
+	}
+
+	node = queue->head;
+	if (node == NULL)
+	{
+        pthread_cond_wait(&(queue->cond), &(queue->lock));
+        node = queue->head;
+    }
+
+	if (node != NULL)
+    {
+		queue->head = node->next;
+		if (queue->head == NULL)
+		{
+			queue->tail = NULL;
+		}
+
+        data = node->data;
+        fast_mblock_free_object(&queue->mblock, node);
+	}
+    else
+    {
+        data = NULL;
+    }
+
+	if ((result=pthread_mutex_unlock(&(queue->lock))) != 0)
+	{
+		logError("file: "__FILE__", line: %d, " \
+			"call pthread_mutex_unlock fail, " \
+			"errno: %d, error info: %s", \
+			__LINE__, result, STRERROR(result));
+	}
+
+	return data;
+}

src/common_blocked_queue.h

@@ -0,0 +1,61 @@
+/**
+* Copyright (C) 2008 Happy Fish / YuQing
+*
+* FastDFS may be copied only under the terms of the GNU General
+* Public License V3, which may be found in the FastDFS source kit.
+* Please visit the FastDFS Home Page http://www.csource.org/ for more detail.
+**/
+
+//common_blocked_queue.h
+
+#ifndef _COMMON_BLOCKED_QUEUE_H
+#define _COMMON_BLOCKED_QUEUE_H
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <pthread.h>
+#include "common_define.h"
+#include "fast_mblock.h"
+
+struct common_blocked_node
+{
+    void *data;
+    struct common_blocked_node *next;
+};
+
+struct common_blocked_queue
+{
+	struct common_blocked_node *head;
+	struct common_blocked_node *tail;
+    struct fast_mblock_man mblock;
+	pthread_mutex_t lock;
+	pthread_cond_t cond;
+};
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+int common_blocked_queue_init_ex(struct common_blocked_queue *queue,
+        const int alloc_elements_once);
+
+#define common_blocked_queue_init(queue)  \
+        common_blocked_queue_init_ex(queue, 1024)
+
+void common_blocked_queue_destroy(struct common_blocked_queue *queue);
+
+static inline void common_blocked_queue_terminate(struct common_blocked_queue *queue)
+{
+     pthread_cond_signal(&(queue->cond));
+}
+
+int common_blocked_queue_push(struct common_blocked_queue *queue, void *data);
+
+void *common_blocked_queue_pop(struct common_blocked_queue *queue);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

src/sched_thread.h

@@ -112,6 +112,7 @@ extern volatile time_t g_current_time;  //the current time
  *  parameters:
  *  	     pScheduleArray: the schedule tasks
  * return: error no, 0 for success, != 0 fail
+ * Note: you should call this function after sched_start
 */
 int sched_add_entries(const ScheduleArray *pScheduleArray);