implement own lock

cmd_in_second.c

@@ -34,6 +34,7 @@ typedef struct cmd_in_second_buffer {
 
 typedef struct cmd_in_second_timer {
     struct timeval* ring;
+    struct timeval last_elem;
     int32_t front;
     int32_t rear;
     int32_t capacity;
@@ -48,6 +49,7 @@ struct cmd_in_second {
     int32_t bulk_limit;
     int32_t log_per_timer;
     state cur_state;
+    pthread_mutex_t lock;
 };
 
 static struct cmd_in_second this;
@@ -61,9 +63,9 @@ static bool is_bulk_cmd()
     }
 
     const struct timeval* front_time = &this.timer.ring[this.timer.front];
-    const struct timeval* rear_time = &this.timer.ring[this.timer.rear];
+    const struct timeval* last_time = &this.timer.last_elem;
 
-    return rear_time->tv_sec - front_time->tv_sec <= 1;
+    return last_time->tv_sec - front_time->tv_sec <= 1;
 }
 
 static void get_whole_cmd(char* whole_cmd)
@@ -196,7 +198,6 @@ static void buffer_add(const logtype* log)
         }
         buffer->front = (buffer->front+1) % buffer->capacity;
     }
-
 }
 
 static void timer_add()
@@ -214,6 +215,7 @@ static void timer_add()
         return;
     };
 
+    timer->last_elem = timer->ring[timer->rear];
     timer->rear = (timer->rear+1) % timer->capacity;
 }
 
@@ -226,7 +228,9 @@ static bool is_cmd_to_log(const char* collection_name, const char* cmd)
 bool cmd_in_second_write(const char* collection_name, const char* cmd,
                          const char* key, const char* client_ip)
 {
+    pthread_mutex_lock(&this.lock);
     if (this.cur_state != ON_LOGGING || !is_cmd_to_log(collection_name, cmd)) {
+        pthread_mutex_unlock(&this.lock);
         return false;
     }
 
@@ -241,38 +245,56 @@ bool cmd_in_second_write(const char* collection_name, const char* cmd,
     buffer_add(&log);
     this.timer.circular_counter = (this.timer.circular_counter+1) % this.log_per_timer;
 
+    pthread_mutex_unlock(&this.lock);
     return true;
 }
 
+void cmd_in_second_init()
+{
+    assert("test");
+    this.cur_state = NOT_STARTED;
+    pthread_mutex_init(&this.lock, NULL);
+
+    this.buffer.front = 0;
+    this.buffer.rear = 0;
+    this.buffer.ring = NULL;
+
+    this.timer.front = 0;
+    this.timer.rear = 0;
+    this.timer.capacity = 0;
+    this.timer.circular_counter = 0;
+    this.timer.ring = NULL;
+}
+
 int32_t cmd_in_second_start(const char* collection_name, const char* cmd,
                             const int32_t bulk_limit)
 {
 
+    pthread_mutex_lock(&this.lock);
+
     if (this.cur_state != NOT_STARTED) {
+        pthread_mutex_unlock(&this.lock);
         return CMD_IN_SECOND_STARTED_ALREADY;
     }
 
     this.bulk_limit = bulk_limit;
 
     this.buffer.capacity = bulk_limit+1;
-    this.buffer.front = 0;
-    this.buffer.rear = 0;
     this.buffer.ring = (logtype*)malloc(this.buffer.capacity * sizeof(logtype));
 
     if (this.buffer.ring == NULL) {
+        pthread_mutex_unlock(&this.lock);
         return CMD_IN_SECOND_NO_MEM;
     }
 
     this.log_per_timer = bulk_limit / 10 + (bulk_limit % 10 != 0);
     this.timer.capacity = this.log_per_timer + 1;
-    this.timer.front = 0;
-    this.timer.rear = 0;
-    this.timer.circular_counter = 0;
 
     this.timer.ring = (struct timeval*)malloc(this.timer.capacity * sizeof(struct timeval));
 
     if (this.timer.ring == NULL) {
         free(this.buffer.ring);
+        pthread_mutex_unlock(&this.lock);
         return CMD_IN_SECOND_NO_MEM;
     }
 
@@ -281,5 +303,7 @@ int32_t cmd_in_second_start(const char* collection_name, const char* cmd,
 
     this.cur_state = ON_LOGGING;
 
+    pthread_mutex_unlock(&this.lock);
+
     return CMD_IN_SECOND_START;
 }

cmd_in_second.h

@@ -1,5 +1,5 @@
-#ifndef __CMD_IN_SECOND_LOG__
-#define __CMD_IN_SECOND_LOG__
+#ifndef __CMD_IN_SECOND_
+#define __CMD_IN_SECOND_
 #endif
 
 #include <stdbool.h>
@@ -10,5 +10,6 @@
 #define CMD_IN_SECOND_STARTED_ALREADY 1
 #define CMD_IN_SECOND_NO_MEM 2
 
+void cmd_in_second_init(void);
 int32_t cmd_in_second_start(const char* collection_name, const char* cmd, const int32_t bulk_limit);
 bool cmd_in_second_write(const char* collection_name, const char* cmd, const char* key, const char* client_ip);