[vhffs-dev] [1314] Fixed race condition when a file is moved right away after being closed for writing / Improved file management

[subversion@xxxxxxxxxxxxx]

Revision: 1314
Author:   gradator
Date:     2008-12-07 02:08:34 +0100 (Sun, 07 Dec 2008)

Log Message:
-----------
Fixed race condition when a file is moved right away after being closed for writing / Improved file management

Modified Paths:
--------------
    trunk/vhffs-fssync/vhffsfssync_master.c


Modified: trunk/vhffs-fssync/vhffsfssync_master.c
===================================================================
--- trunk/vhffs-fssync/vhffsfssync_master.c	2008-12-04 22:46:43 UTC (rev 1313)
+++ trunk/vhffs-fssync/vhffsfssync_master.c	2008-12-07 01:08:34 UTC (rev 1314)
@@ -56,11 +56,6 @@
 // Not used yet: IN_ATTRIB, IN_DELETE_SELF, IN_MODIFY, IN_MOVE_SELF
 // Will never be used: IN_ACCESS, IN_OPEN, IN_CLOSE_NOWRITE
 
-// the maximum number of files simultaneously opened
-// huge values offer better performance
-// actually it is  MAX = VHFFSFSSYNC_MAX_OPENFILES + files currently uploaded
-#define VHFFSFSSYNC_MAX_OPENFILES 512
-
 // each monitor entry is associated with a path, we need to keep it to compute the path
 //char **vhffsfssync_wd_to_path = NULL;
 //int vhffsfssync_wd_to_path_len = 0;  // number of allocated paths
@@ -81,8 +76,6 @@
 };
 static struct vhffsfssync_cookie vhffsfssync_cookie;
 
-int vhffsfssync_openfiles;
-
 GHashTable *vhffsfssync_missedfiles;
 
 // protos
@@ -171,28 +164,36 @@
 	 - sizeof(ssize_t) ];
 } vhffsfssync_net_message_data;
 
+/* net - filehandle */
+typedef struct {
+	FILE *file_stream;
+	char *file_pathname;
+	off_t file_size;
+	int ref;
+} vhffsfssync_net_file;
+
+GHashTable *vhffsfssync_net_files;
+
+
 /* message - file */
 typedef struct {
 	__VHFFSFSSYNC_NET_MESSAGE_COMMON(file_);
-	FILE *file_stream;
+	vhffsfssync_net_file *file;
 	off_t file_offset;
-	off_t file_size;
 	ssize_t file_chunksize;
 	ssize_t file_chunkcur;
-	char *file_pathname;
 
 	/* pad to size of `vhffsfssync_net_message'  */
 	unsigned char sin_zero[ sizeof(vhffsfssync_net_message)
 	 - __VHFFSFSSYNC_NET_MESSAGE_COMMON_SIZE
-	 - sizeof(FILE*)
+	 - sizeof(vhffsfssync_net_file*)
 	 - sizeof(off_t)
-	 - sizeof(off_t)
 	 - sizeof(ssize_t)
-	 - sizeof(ssize_t)
-	 - sizeof(char*) ];
+	 - sizeof(ssize_t) ];
 } vhffsfssync_net_message_file;
 
 
+
 // protos
 void vhffsfssync_net_conn_disable(vhffsfssync_conn *conn);
 void vhffsfssync_net_conn_destroy(vhffsfssync_conn *conn);
@@ -208,8 +209,8 @@
 void vhffsfssync_net_broadcast_event(char *data, uint32_t priority);
 int vhffsfssync_net_send_file(vhffsfssync_conn *conn, char *pathname);
 void vhffsfssync_net_destroy_file(vhffsfssync_conn *conn, vhffsfssync_net_message_file *filemsg);
-FILE *vhffsfssync_net_file_open(vhffsfssync_conn *conn, const char *pathname, const char *mode);
-int vhffsfssync_net_file_close(vhffsfssync_conn *conn, FILE *stream);
+vhffsfssync_net_file *vhffsfssync_net_file_open(vhffsfssync_conn *conn, const char *pathname, const char *mode);
+int vhffsfssync_net_file_close(vhffsfssync_conn *conn, vhffsfssync_net_file *file);
 int vhffsfssync_net_remove_file(vhffsfssync_conn *conn, char *pathname);
 void vhffsfssync_net_broadcast_file(char *pathname);
 int vhffsfssync_net_send(vhffsfssync_conn *conn);
@@ -401,30 +402,18 @@
 // prototype is simple, files are always the lowest of the lowest priority messages
 int vhffsfssync_net_send_file(vhffsfssync_conn *conn, char *pathname)  {
 	vhffsfssync_net_message_file *msg;
-	struct stat st;
 	uint32_t maxprio = -1;  // 4294967295
-	FILE *stream;
+	vhffsfssync_net_file *file;
 
-	if(!conn || !pathname) {
+	if(!conn || !pathname)
 		return -1;
-	}
 
-	if( lstat(pathname, &st) < 0 ) {
-		fprintf(stderr, "lstat() failed on %s: %s\n", pathname, strerror(errno));
+	file = vhffsfssync_net_file_open(conn, pathname, "r");
+	if(!file) {
+		g_hash_table_insert(vhffsfssync_missedfiles, g_strdup(pathname), "");
 		return -1;
 	}
 
-	/* only copy regular files */
-	if(! S_ISREG(st.st_mode) ) {
-		fprintf(stderr, "%s is not a regular file\n", pathname);
-		return -1;
-	}
-
-	stream = vhffsfssync_net_file_open(conn, pathname, "r");
-	if(!stream) {
-		return -1;
-	}
-
 	// if the file is being sent, cancel it
 	vhffsfssync_net_remove_file(conn, pathname);
 	//printf("%d SENDING FILE %s\n", conn->fd, pathname);
@@ -432,11 +421,9 @@
 
 	// the size of the file is the priority (small files are sent with more priority)
 	// but don't set the priority too low, low value can be used for anything else
-	msg = (vhffsfssync_net_message_file*)vhffsfssync_net_new_message(conn, VHFFSFSSYNC_NET_MESSAGE_FILE, MAX(MIN(st.st_size, maxprio),1000) );
-	msg->file_stream = stream;
+	msg = (vhffsfssync_net_message_file*)vhffsfssync_net_new_message(conn, VHFFSFSSYNC_NET_MESSAGE_FILE, MAX(MIN(file->file_size, maxprio),1000) );
+	msg->file = file;
 	msg->file_offset = 0;
-	msg->file_pathname = strdup(pathname);
-	msg->file_size = st.st_size;
 	msg->file_chunksize = -1;
 	msg->file_chunkcur = 0;
 	conn->messages = g_list_insert_sorted(conn->messages, msg, vhffsfssync_net_message_insert_compare);
@@ -460,63 +447,74 @@
 		}
 	}
 
-	vhffsfssync_net_send_event(conn, g_strdup_printf("close%c%s%c", '\0', filemsg->file_pathname, '\0') , VHFFSFSSYNC_NET_PRIO_MEDIUM);
-	vhffsfssync_net_file_close(conn, filemsg->file_stream);
-	free(filemsg->file_pathname);
+	vhffsfssync_net_send_event(conn, g_strdup_printf("close%c%s%c", '\0', filemsg->file->file_pathname, '\0') , VHFFSFSSYNC_NET_PRIO_MEDIUM);
+	vhffsfssync_net_file_close(conn, filemsg->file);
 	free(filemsg);
 }
 
 
-FILE *vhffsfssync_net_file_open(vhffsfssync_conn *conn, const char *pathname, const char *mode)  {
+vhffsfssync_net_file *vhffsfssync_net_file_open(vhffsfssync_conn *conn, const char *pathname, const char *mode)  {
 	FILE *stream;
+	vhffsfssync_net_file *file;
+	struct stat st;
 
 	if(!conn || !pathname || !mode)
 		return NULL;
 
-	stream = fopen(pathname, mode);
-	if(!stream) {
-		fprintf(stderr, "fopen() failed on %s: %s\n", pathname, strerror(errno));
-		return stream;
+	file = g_hash_table_lookup(vhffsfssync_net_files, pathname);
+	if(file) {
+		file->ref++;
+		// update size of file (it might have changed)
+		fstat(fileno(file->file_stream), &st);
+		file->file_size = st.st_size;
+		return file;
 	}
 
-	vhffsfssync_openfiles++;
-	// do we need to free a file fd ?
-	if(vhffsfssync_openfiles > VHFFSFSSYNC_MAX_OPENFILES) {
-		GList *msgs;
-		fprintf(stderr, "Maximum number of opened files reached, consider adding more\n");
-		// we prefer to fclose() the fd of the msg with the lowest priority
-		for(msgs = g_list_last(conn->messages) ; msgs ; )  {
-			vhffsfssync_net_message *msg = msgs->data;
-			msgs = g_list_previous(msgs);
+	if( lstat(pathname, &st) < 0 ) {
+		fprintf(stderr, "lstat() failed on %s: %s\n", pathname, strerror(errno));
+		return NULL;
+	}
 
-			if(msg->msg_family == VHFFSFSSYNC_NET_MESSAGE_FILE)  {
-				vhffsfssync_net_message_file *filemsg = (vhffsfssync_net_message_file*)msg;
+	/* only copy regular files */
+	if(! S_ISREG(st.st_mode) ) {
+		fprintf(stderr, "%s is not a regular file\n", pathname);
+		return NULL;
+	}
 
-				if(filemsg->file_stream  &&  !filemsg->file_offset  &&  filemsg->file_chunksize < 0) {
-					vhffsfssync_net_file_close(conn, filemsg->file_stream);
-					filemsg->file_stream = NULL;
-					break;
-				}
-			}
-		}
+	stream = fopen(pathname, mode);
+	if(!stream) {
+		fprintf(stderr, "fopen() failed on %s: %s\n", pathname, strerror(errno));
+		return NULL;
 	}
 
-	return stream;
+	file = malloc(sizeof(vhffsfssync_net_file));
+	file->file_stream = stream;
+	file->file_pathname = strdup(pathname);
+	file->file_size = st.st_size;
+	file->ref = 1;
+	g_hash_table_insert(vhffsfssync_net_files, file->file_pathname, file);
+	return file;
 }
 
 
-int vhffsfssync_net_file_close(vhffsfssync_conn *conn, FILE *stream)  {
+int vhffsfssync_net_file_close(vhffsfssync_conn *conn, vhffsfssync_net_file *file)  {
 	int r;
 
-	if(!stream)
+	if(!file)
 		return -1;
 
-	r = fclose(stream);
-	if( r ) {
+	file->ref--;
+	if(file->ref)
+		return 0;
+
+	g_hash_table_remove(vhffsfssync_net_files, file->file_pathname);
+
+	r = fclose(file->file_stream);
+	if( r )
 		fprintf(stderr, "fclose() failed: %s\n", strerror(errno));
-		return r;
-	}
-	vhffsfssync_openfiles--;
+
+	free(file->file_pathname);
+	free(file);
 	return r;
 }
 
@@ -540,7 +538,7 @@
 
 		if(msg->msg_family == VHFFSFSSYNC_NET_MESSAGE_FILE)  {
 			vhffsfssync_net_message_file *filemsg = (vhffsfssync_net_message_file*)msg;
-			if(!strcmp(filemsg->file_pathname, pathname)) {
+			if(filemsg->file && !strcmp(filemsg->file->file_pathname, pathname)) {
 				//printf("%d CANCELLING %s\n", conn->fd, pathname);
 				vhffsfssync_net_destroy_file(conn, filemsg);
 			}
@@ -618,23 +616,15 @@
 
 			filemsg = (vhffsfssync_net_message_file*)msg;
 #if DEBUG_NET
-			printf("    file: %s, offset = %lld, size = %lld\n", filemsg->file_pathname, (long long int)filemsg->file_offset, (long long int)filemsg->file_size);
+			printf("    file: %s, offset = %lld, size = %lld\n", filemsg->file->file_pathname, (long long int)filemsg->file_offset, (long long int)filemsg->file->file_size);
 #endif
 			/* new chunk */
 			if(filemsg->file_chunksize < 0)  {
-				if(!filemsg->file_stream) {
-					filemsg->file_stream = vhffsfssync_net_file_open(conn, filemsg->file_pathname, "r");
-					if(!filemsg->file_stream) {
-						vhffsfssync_net_destroy_file(conn, filemsg);
-						break;
-					}
-				}
-
-				lentowrite = filemsg->file_chunksize = MIN(VHFFSFSSYNC_NET_MESSAGE_FILE_CHUNK, filemsg->file_size - filemsg->file_offset);
+				lentowrite = filemsg->file_chunksize = MIN(VHFFSFSSYNC_NET_MESSAGE_FILE_CHUNK, filemsg->file->file_size - filemsg->file_offset);
 				filemsg->file_chunkcur = 0;
 				// we need to make sure that the chunk will not be truncated, we set the current message to the highest priority
 				msg->msg_priority = 1;
-				vhffsfssync_net_send_event(conn, g_strdup_printf("write%c%s%c%lld%c%ld%c", '\0', filemsg->file_pathname, '\0', (long long int)filemsg->file_offset, '\0', (long int)filemsg->file_chunksize, '\0') , 0);
+				vhffsfssync_net_send_event(conn, g_strdup_printf("write%c%s%c%lld%c%ld%c", '\0', filemsg->file->file_pathname, '\0', (long long int)filemsg->file_offset, '\0', (long int)filemsg->file_chunksize, '\0') , 0);
 				// we need to reset here in order to consider the new priorities
 				continue;
 			}
@@ -644,7 +634,7 @@
 			}
 
 			/* try to send the file */
-			written = sendfile(conn->fd, fileno(filemsg->file_stream), &filemsg->file_offset, lentowrite);
+			written = sendfile(conn->fd, fileno(filemsg->file->file_stream), &filemsg->file_offset, lentowrite);
 			if(written < 0) {
 				switch(errno)  {
 					case EAGAIN:
@@ -655,7 +645,7 @@
 						full = TRUE;
 						break;
 					default:
-						fprintf(stderr, "sendfile() failed from file %s to socket %d: %s\n", filemsg->file_pathname, conn->fd, strerror(errno));
+						fprintf(stderr, "sendfile() failed from file %s to socket %d: %s\n", filemsg->file->file_pathname, conn->fd, strerror(errno));
 						vhffsfssync_net_conn_disable(conn);
 				}
 			}
@@ -666,7 +656,7 @@
 				filemsg->file_chunkcur += written;
 
 				/* end of file or file completed */
-				if( written == 0 || filemsg->file_offset == filemsg->file_size )  {
+				if( written == 0 || filemsg->file_offset == filemsg->file->file_size )  {
 					vhffsfssync_net_destroy_file(conn, filemsg);
 				}
 
@@ -683,7 +673,7 @@
 					filemsg->file_chunkcur = 0;
 
 					// reschedule this file to a nicer priority
-					msg->msg_priority = MAX(MIN(filemsg->file_size - filemsg->file_offset, maxprio),1000);
+					msg->msg_priority = MAX(MIN(filemsg->file->file_size - filemsg->file_offset, maxprio),1000);
 					conn->messages = g_list_remove(conn->messages, msg);
 					conn->messages = g_list_insert_sorted(conn->messages, msg, vhffsfssync_net_message_insert_compare);
 				}
@@ -1477,6 +1467,7 @@
 				vhffsfssync_net_broadcast_event( g_strdup_printf("move%c%s%c%s%c", '\0', vhffsfssync_cookie.from, '\0', pathname, '\0') , VHFFSFSSYNC_NET_PRIO_MEDIUM);
 			}
 			else {
+				vhffsfssync_manage_event_remove(inotifyfd, vhffsfssync_cookie.from);
 				vhffsfssync_manage_event_create(inotifyfd, pathname, TRUE);
 				g_hash_table_remove(vhffsfssync_missedfiles, vhffsfssync_cookie.from);
 			}
@@ -1566,7 +1557,6 @@
 	vhffsfssync_path_to_wd = g_hash_table_new_full(g_str_hash, g_str_equal, NULL, NULL);
 	vhffsfssync_cookie.id = 0;
 	vhffsfssync_cookie.from = NULL;
-	vhffsfssync_openfiles = 0;
 	vhffsfssync_missedfiles = g_hash_table_new_full(g_str_hash, g_str_equal, g_free, NULL);
 
 	inotifyfd = inotify_init();
@@ -1586,6 +1576,7 @@
 
 	/* -- network stuff -- */
 	vhffsfssync_conns = NULL;
+	vhffsfssync_net_files = g_hash_table_new_full(g_str_hash, g_str_equal, NULL, NULL);
 
 	signal(SIGPIPE, SIG_IGN);