add functions: fc_queue_push_with_check and fc_queue_peek

HISTORY

@@ -1,5 +1,5 @@
 
-Version 1.70  2023-09-25
+Version 1.70  2023-09-30
   * get full mac address of infiniband NIC under Linux
   * struct fast_task_info add field conn for RDMA connection
   * server_id_func.[hc]: support communication type
@@ -7,6 +7,7 @@ Version 1.70  2023-09-25
   * nio thread data support busy_polling_callback
   * connection_pool.[hc] support thread local for performance
   * struct fast_task_info support send and recv double buffers
+  * add functions: fc_queue_push_with_check and fc_queue_peek
 
 Version 1.69  2023-08-05
   * bugfixed: array_allocator_alloc MUST init the array

src/fc_queue.c

@@ -50,10 +50,53 @@ void fc_queue_push_ex(struct fc_queue *queue, void *data, bool *notify)
         *notify = false;
 	}
 	queue->tail = data;
-
     PTHREAD_MUTEX_UNLOCK(&queue->lcp.lock);
 }
 
+static inline bool fc_queue_exists(struct fc_queue *queue, void *data)
+{
+    void *current;
+    if (queue->head == NULL) {
+        return false;
+    }
+
+    current = queue->head;
+    do {
+        if (current == data) {
+            return true;
+        }
+        current = FC_QUEUE_NEXT_PTR(queue, current);
+    } while (current != NULL);
+
+    return false;
+}
+
+int fc_queue_push_with_check_ex(struct fc_queue *queue,
+        void *data, bool *notify)
+{
+    int result;
+
+    PTHREAD_MUTEX_LOCK(&queue->lcp.lock);
+    if (fc_queue_exists(queue, data)) {
+        result = EEXIST;
+        *notify = false;
+    } else {
+        result = 0;
+        FC_QUEUE_NEXT_PTR(queue, data) = NULL;
+        if (queue->tail == NULL) {
+            queue->head = data;
+            *notify = true;
+        } else {
+            FC_QUEUE_NEXT_PTR(queue, queue->tail) = data;
+            *notify = false;
+        }
+        queue->tail = data;
+    }
+    PTHREAD_MUTEX_UNLOCK(&queue->lcp.lock);
+
+    return result;
+}
+
 void *fc_queue_pop_ex(struct fc_queue *queue, const bool blocked)
 {
 	void *data;

src/fc_queue.h

@@ -68,6 +68,8 @@ static inline void fc_queue_terminate_all(
 
 //notify by the caller
 void fc_queue_push_ex(struct fc_queue *queue, void *data, bool *notify);
+int fc_queue_push_with_check_ex(struct fc_queue *queue,
+        void *data, bool *notify);
 
 static inline void fc_queue_push(struct fc_queue *queue, void *data)
 {
@@ -79,6 +81,19 @@ static inline void fc_queue_push(struct fc_queue *queue, void *data)
     }
 }
 
+static inline int fc_queue_push_with_check(struct fc_queue *queue, void *data)
+{
+    int result;
+    bool notify;
+
+    result = fc_queue_push_with_check_ex(queue, data, &notify);
+    if (notify) {
+        pthread_cond_signal(&(queue->lcp.cond));
+    }
+
+    return result;
+}
+
 static inline void fc_queue_push_silence(struct fc_queue *queue, void *data)
 {
     bool notify;
@@ -171,6 +186,16 @@ static inline int fc_queue_count(struct fc_queue *queue)
     return count;
 }
 
+static inline void *fc_queue_peek(struct fc_queue *queue)
+{
+    void *data;
+
+    pthread_mutex_lock(&queue->lcp.lock);
+    data = queue->head;
+    pthread_mutex_unlock(&queue->lcp.lock);
+    return data;
+}
+
 void *fc_queue_timedpop(struct fc_queue *queue,
         const int timeout, const int time_unit);