Changeset 381

Timestamp:

03/21/05 01:07:30 (4 years ago)

Author:

azrael

Message:

Added skeleton for the new batch commit plugin
functions.

Added a new signal to the members "committed_all" that
will be called by the engine when all changes to be written
have been sent to a member.

Location:

trunk

Files:

• Makefile.am (modified) (1 diff)
• opensync/opensync_format_internals.h (modified) (1 diff)
• opensync/opensync_member.c (modified) (3 diffs)
• opensync/opensync_member.h (modified) (1 diff)
• opensync/opensync_plugin.h (modified) (1 diff)
• osengine/osengine_client.c (modified) (3 diffs)
• osengine/osengine_client_internals.h (modified) (1 diff)
• osengine/osengine_engine.c (modified) (1 diff)
• osengine/osengine_engine_internals.h (modified) (1 diff)

trunk/Makefile.am

@@ -18 +18 @@
 EXTRA_DIST = \
         autogen.sh \
-        buildtest.sh \
         plugins \
         opensync-1.0.pc.in \
-        osengine-1.0.pc.in
+        osengine-1.0.pc.in \
+        misc/buildtest.sh \
+        misc/doxygen.css
 
 dist-hook:

trunk/opensync/opensync_format_internals.h

@@ -84 +84 @@
         char *extension_name;
         struct OSyncObjTypeSink *objtype_sink;
+        GList *commit_changes;
+        GList *commit_contexts;
 } OSyncObjFormatSink;
 

trunk/opensync/opensync_member.c

@@ -563 +563 @@
         g_assert(change);
 
+        OSyncFormatEnv *env = osync_member_get_format_env(member);
         OSyncContext *context = osync_context_new(member);
         context->callback_function = function;
         context->calldata = user_data;
-
-        OSyncFormatEnv *env = osync_member_get_format_env(member);
+        
         
         OSyncObjType *type = change->objtype;
@@ -618 +618 @@
                 osync_debug("OSYNC", 2, "Comparing change %s with sink %s", osync_change_get_objformat(change)->name, fmtsink->format ? fmtsink->format->name : "None");
                 if (fmtsink->format == osync_change_get_objformat(change)) {
-                        // Send the change
-                        fmtsink->functions.commit_change(context, change);
-                        osync_trace(TRACE_EXIT, "%s", __func__);
-                        return;
+                        if (fmtsink->functions.batch_commit) {
+                                //Append to the stored changes
+                                fmtsink->commit_changes = g_list_append(fmtsink->commit_changes, change);
+                                fmtsink->commit_contexts = g_list_append(fmtsink->commit_contexts, context);
+                                osync_trace(TRACE_EXIT, "%s: Waiting for batch processing", __func__);
+                                return;
+                        } else {
+                                // Send the change
+                                fmtsink->functions.commit_change(context, change);
+                                osync_trace(TRACE_EXIT, "%s", __func__);
+                                return;
+                        }
                 }
         }
@@ -627 +635 @@
         osync_context_report_error(context, OSYNC_ERROR_CONVERT, "Unable to send changes");
         osync_trace(TRACE_EXIT_ERROR, "%s: Unable to find a sink", __func__);
+}
+
+void osync_member_committed_all(OSyncMember *member)
+{
+        osync_trace(TRACE_ENTRY, "%s(%p)", __func__, member);
+
+        GList *f;
+        for (f = member->format_sinks; f; f = f->next) {
+                OSyncObjFormatSink *fmtsink = f->data;
+                osync_debug("OSYNC", 2, "Sending committed all to sink %s", fmtsink->format ? fmtsink->format->name : "None");
+
+                OSyncFormatFunctions functions = fmtsink->functions;
+
+                if (functions.batch_commit) {
+                        GList *o = fmtsink->commit_contexts;
+                        GList *c = NULL;
+                        
+                        int i = 0;
+                        OSyncChange **changes = g_malloc0(sizeof(OSyncChange *) * (g_list_length(fmtsink->commit_changes) + 1));
+                        OSyncContext **contexts = g_malloc0(sizeof(OSyncContext *) * (g_list_length(fmtsink->commit_changes) + 1));;
+                        
+                        for (c = fmtsink->commit_changes; c && o; c = c->next) {
+                                OSyncChange *change = c->data;
+                                OSyncContext *context = o->data;
+                                
+                                changes[i] = change;
+                                contexts[i] = context;
+                                
+                                i++;
+                                o = o->next;
+                        }
+                        
+                        changes[i] = NULL;
+                        contexts[i] = NULL;
+                        functions.batch_commit(member->plugindata, contexts, changes);
+                        
+                        g_free(changes);
+                        g_free(contexts);
+                        
+                        g_list_free(fmtsink->commit_changes);
+                        fmtsink->commit_changes = NULL;
+                        
+                        g_list_free(fmtsink->commit_contexts);
+                        fmtsink->commit_contexts = NULL;
+                } else if (functions.committed_all) {
+                        functions.committed_all(member->plugindata);
+                }
+        }
+        
+        osync_trace(TRACE_EXIT, "%s", __func__);
 }
 

trunk/opensync/opensync_member.h

@@ -40 +40 @@
 void osync_member_get_changeinfo(OSyncMember *member, OSyncEngCallback function, void *user_data);
 void osync_member_read_change(OSyncMember *member, OSyncChange *change, OSyncEngCallback function, void *user_data);
+void osync_member_committed_all(OSyncMember *member);
 
 void *osync_member_call_plugin(OSyncMember *member, const char *function, void *data, OSyncError **error);

trunk/opensync/opensync_plugin.h

@@ -36 +36 @@
 typedef struct OSyncFormatFunctions {
         osync_bool (* commit_change) (OSyncContext *, OSyncChange *);
+        void (* committed_all) (void *);
+        void (* batch_commit) (void *, OSyncContext **, OSyncChange **);
         osync_bool (* access) (OSyncContext *, OSyncChange *);
         void (* read) (OSyncContext *, OSyncChange *);

trunk/osengine/osengine_client.c

@@ -42 +42 @@
         client->fl_done = osync_flag_new(NULL);
         client->fl_finished = osync_flag_new(engine->cmb_finished);
-        //client->fl_reset = osync_flag_new(engine->cmb_reset);
-        
+        
+        client->cmb_written = osync_comb_flag_new(FALSE, FALSE);
+        osync_flag_set_pos_trigger(client->cmb_written, (MSyncFlagTriggerFunc)send_committed_all, client, engine);
+                
     return client;
 }
@@ -61 +63 @@
         osync_flag_free(client->fl_done);
         osync_flag_free(client->fl_finished);
+        osync_flag_free(client->cmb_written);
 
         g_free(client);
@@ -171 +174 @@
         }
 
+        if (itm_message_is_methodcall(message, "COMMITTED_ALL")) {
+                osync_member_committed_all(client->member);
+                osync_trace(TRACE_EXIT, "client_message_handler");
+                return;
+        }
+        
         if (itm_message_is_methodcall(message, "READ_CHANGE")) {
                 OSyncChange *change = itm_message_get_data(message, "change");

trunk/osengine/osengine_client_internals.h

@@ -15 +15 @@
         MSyncFlag *fl_done;
         MSyncFlag *fl_finished;
+        MSyncFlag *cmb_written;
         GThread *thread;
         GMainContext *context;

trunk/osengine/osengine_engine.c

@@ -348 +348 @@
 }
 
+void send_committed_all(OSyncClient *target, OSyncEngine *sender)
+{
+        osync_flag_changing(target->fl_done);
+        ITMessage *message = itm_message_new_signal(sender, "COMMITTED_ALL");
+        itm_queue_send(target->incoming, message);
+}
+
 void send_disconnect(OSyncClient *target, OSyncEngine *sender)
 {

trunk/osengine/osengine_engine_internals.h

@@ -70 +70 @@
 void send_read_change(OSyncEngine *sender, OSyncMappingEntry *entry);
 void send_engine_changed(OSyncEngine *engine);
+void send_committed_all(OSyncClient *target, OSyncEngine *sender);