db_flatstore: Fix missing CDRs due to 'flat_rotate' race conditions

This commit simply adds R/W locking around the 'flat_rotate' MI command,
and all the SIP worker processes.  The race condition was as follows:

* SIP worker checks the 'flat_rotate' stamp and proceeds to write CDR
* 'flat_rotate' is ran (timestamp++)
* external process scans & fully reads the rotated file (with deletion
   to follow later as well, but this is irrelevant)
* SIP worker finally writes the CDR using flushed/vector'ed I/O
   (too late at this point, CDR is forever lost)

(cherry picked from commit d3b1463)

modules/db_flatstore/flat_mi.c

@@ -23,10 +23,18 @@
 #include "flatstore_mod.h"
 #include "flat_mi.h"
 
+/* helps ensure that there are no more pending writes on the old (rotated)
+ * files, thus avoiding race conditions with external readers */
+rw_lock_t *rotate_lock;
+
 mi_response_t *mi_flat_rotate_cmd(const mi_params_t *params,
 								struct mi_handler *async_hdl)
 {
-	*flat_rotate = time(0);
+	time_t now = time(NULL);
+
+	lock_start_write(rotate_lock);
+	*flat_rotate = now;
+	lock_stop_write(rotate_lock);
 
 	return init_mi_result_ok();
 }

modules/db_flatstore/flat_mi.h

@@ -24,9 +24,12 @@
 #define _FLATSTORE_MI_H_
 
 #include "../../mi/mi.h"
+#include "../../rw_locking.h"
 
 #define MI_FLAT_ROTATE "flat_rotate"
 
+extern rw_lock_t *rotate_lock;
+
 mi_response_t *mi_flat_rotate_cmd(const mi_params_t *params,
 								struct mi_handler *async_hdl);
 

modules/db_flatstore/flatstore.c

@@ -36,6 +36,7 @@
 #include "../../dprint.h"
 #include "flat_pool.h"
 #include "flat_con.h"
+#include "flat_mi.h"
 #include "flatstore_mod.h"
 #include "flatstore.h"
 
@@ -336,19 +337,21 @@ int flat_db_insert(const db_con_t* h, const db_key_t* k, const db_val_t* v,
 	str aux;
 	char * begin = flat_iov_buf.s;
 
+	lock_start_read(rotate_lock);
+
 	if (local_timestamp < *flat_rotate) {
 		flat_rotate_logs();
 		local_timestamp = *flat_rotate;
 	}
 
 	if ( !h || !CON_TAIL(h) || (f=CON_FILE(h))==NULL ) {
 		LM_ERR("uninitialized connection\n");
-		return -1;
+		goto out_err;
 	}
 
 	if (flat_prepare_iovec(n) < 0) {
 		LM_ERR("cannot insert row\n");
-		return -1;
+		goto out_err;
 	}
 
 	FLAT_LOCK(f);
@@ -427,7 +430,7 @@ int flat_db_insert(const db_con_t* h, const db_key_t* k, const db_val_t* v,
 
 	if (auxl < 0) {
 		LM_ERR("unable to write to file: %s - %d\n", strerror(errno), errno);
-		return -1;
+		goto out_err;
 	}
 
 	/* XXX does this make sense any more? */
@@ -436,6 +439,10 @@ int flat_db_insert(const db_con_t* h, const db_key_t* k, const db_val_t* v,
 	}
 	FLAT_UNLOCK(f);
 
-
+	lock_stop_read(rotate_lock);
 	return 0;
+
+out_err:
+	lock_stop_read(rotate_lock);
+	return -1;
 }

modules/db_flatstore/flatstore_mod.c

@@ -150,6 +150,11 @@ static int mod_init(void)
 	*flat_rotate = time(0);
 	local_timestamp = *flat_rotate;
 
+	if (!(rotate_lock = lock_init_rw())) {
+		LM_ERR("oom\n");
+		return -1;
+	}
+
 	/* parse prefix and suffix */
 	if (flat_suffix_s.s && (flat_suffix_s.len=strlen(flat_suffix_s.s))!=0) {
 		if (pv_parse_format(&flat_suffix_s, &flat_suffix) < 0) {