Comments update

Author: renukamanavalan

common/events.cpp

@@ -1,36 +1,8 @@
 #include "events_pi.h"
 
-/*
- * The uuid_unparse() function converts the supplied UUID uu from
- * the binary representation into a 36-byte string (plus trailing
- * '\0') of the form 1b4e28ba-2fa1-11d2-883f-0016d3cca427 and stores
- * this value in the character string pointed to by out. 
- */
-#define UUID_STR_SIZE 40
-
-/*
- * Publisher use echo service and subscriber uses cache service
- * The eventd process runs this service, which could be down
- * All service interactions being async, a timeout is required
- * not to block forever on read.
- *
- * The unit is in milliseconds in sync with ZMQ_RCVTIMEO of
- * zmq_setsockopt.
- *
- * Publisher uses more to shadow async connectivity from PUB to 
- * XSUB end point of eventd's proxy. Hene have a less value.
- *
- * Subscriber uses it for cache management and here we need a
- * longer timeout, to handle slow proxy. This timeout value's only
- * impact could be subscriber process trying to terminate.
- */
-
-#define EVENTS_SERVICE_TIMEOUT_MS_PUB 200
-#define EVENTS_SERVICE_TIMEOUT_MS_SUB 2000
-
 /*
  *  Track created publishers to avoid duplicates
- *  As we track missed event count by publishing instances, avoiding
+ *  As receivers track missed event count by publishing instances, avoiding
  *  duplicates helps reduce load.
  */
 
@@ -50,27 +22,33 @@ int EventPublisher::init(const string event_source)
     int rc = zmq_connect (sock, get_config(XSUB_END_KEY).c_str());
     RET_ON_ERR(rc == 0, "Publisher fails to connect %s", get_config(XSUB_END_KEY).c_str());
 
-    // REQ socket is connected and a message is sent & received, more to
-    // ensure PUB socket had enough time to establish connection.
-    // Any message published before connection establishment is dropped.
-    //
     /*
      * Event service could be down. So have a timeout.
      * 
      */
     rc = m_event_service.init_client(m_zmq_ctx, EVENTS_SERVICE_TIMEOUT_MS_PUB);
     RET_ON_ERR (rc == 0, "Failed to init event service");
 
+    /*
+     * REQ socket is connected and a message is sent & received, more to
+     * ensure PUB socket had enough time to establish connection.
+     * Any message published before connection establishment is dropped.
+     * NOTE: We don't wait for response here, but read it upon first publish
+     * If the publisher init happened early at the start by caller, by the
+     * the first event is published, the echo response will be available locally.
+     */
     rc = m_event_service.echo_send("hello");
     RET_ON_ERR (rc == 0, "Failed to echo send in event service");
 
     m_event_source = event_source;
 
+    {
     uuid_t id;
     char uuid_str[UUID_STR_SIZE];
     uuid_generate(id);
     uuid_unparse(id, uuid_str);
     m_runtime_id = string(uuid_str);
+    }
 
     m_socket = sock;
 out:
@@ -104,9 +82,12 @@ EventPublisher::publish(const string tag, const event_params_t *params)
          * as provided in publisher init.
          */
         m_event_service.echo_receive(s);
+
+        /* Close it as we don't need it anymore */
         m_event_service.close_service();
     }
 
+    /* Check for timestamp in params. If not, provide it. */
     string param_str;
     event_params_t evt_params;
     if (params != NULL) {
@@ -120,6 +101,7 @@ EventPublisher::publish(const string tag, const event_params_t *params)
         evt_params[event_ts_param] = get_timestamp();
         params = &evt_params;
     }
+
     rc = serialize(*params, param_str);
     RET_ON_ERR(rc == 0, "failed to serialize params %s",
             map_to_str(*params).c_str());
@@ -232,9 +214,12 @@ EventSubscriber::~EventSubscriber()
             rc = zmq_message_read(m_socket, ZMQ_DONTWAIT, source, evt_data);
             if (rc == -1) {
                 if (zerrno == EAGAIN) {
-                    rc = 0;
+                    /* Try again after a small pause */
+                    this_thread::sleep_for(chrono::milliseconds(10));
+                }
+                else {
+                    break;
                 }
-                break;
             }
             serialize(evt_data, evt_str);
             events.push_back(evt_str);
@@ -315,10 +300,12 @@ EventSubscriber::prune_track()
 {
     map<time_t, vector<runtime_id_t> > lst;
 
+    /* Sort entries by last touched time */
     for(const auto e: m_track) {
         lst[e.second.epoch_secs].push_back(e.first);
     }
 
+    /* By default it walks from lowest value / earliest timestamp */
     map<time_t, vector<runtime_id_t> >::const_iterator itc = lst.begin();
     for(; (itc != lst.end()) && (m_track.size() > MAX_PUBLISHERS_COUNT); ++itc) {
         for (const auto r: itc->second) {

common/events_pi.h

@@ -111,8 +111,7 @@ class EventSubscriber : public events_base
         event_service m_event_service;
 
         /*
-         * Set to true, upon cache read returning non zero count of events
-         * implying more to read.
+         * Set to true, if there may be more events to read from cache.
          */
         bool m_cache_read;
 
@@ -157,4 +156,32 @@ class EventSubscriber : public events_base
  */
 #define CACHE_DRAIN_IN_MILLISECS 1000
 
+/*
+ * The uuid_unparse() function converts the supplied UUID uu from
+ * the binary representation into a 36-byte string (plus trailing
+ * '\0') of the form 1b4e28ba-2fa1-11d2-883f-0016d3cca427 and stores
+ * this value in the character string pointed to by out. 
+ */
+#define UUID_STR_SIZE 40
+
+/*
+ * Publisher use echo service and subscriber uses cache service
+ * The eventd process runs this service, which could be down
+ * All service interactions being async, a timeout is required
+ * not to block forever on read.
+ *
+ * The unit is in milliseconds in sync with ZMQ_RCVTIMEO of
+ * zmq_setsockopt.
+ *
+ * Publisher uses more to shadow async connectivity from PUB to 
+ * XSUB end point of eventd's proxy. Hence have a smaller value.
+ *
+ * Subscriber uses it for cache management and here we need a
+ * longer timeout, to handle slow proxy. This timeout value's only
+ * impact could be when subscriber process is trying to terminate.
+ */
+
+#define EVENTS_SERVICE_TIMEOUT_MS_PUB 200
+#define EVENTS_SERVICE_TIMEOUT_MS_SUB 2000
+
 #endif /* !_EVENTS_PI_H */ 

common/events_service.cpp

@@ -1,25 +1,30 @@
 #include "events_service.h"
 
 /*
- * For brainstorming, if helpful
- * The cache messages are read in either direction
- * Upon start, the caller gives a set of events read for about 2 seconds
- * in non-blocking mode to give it as start stock.
+ * Cache management
  *
- * Upon cache stop, events collected MAX over a time is read by the caller.
- * 
- * These messages are currently provided as vector list of strings.
- * As cache start provided a small subset, it is given as part of start request
- * Since cache read can be too many, multiple cache_Read requests are made
- * until no more and each returns a subset as vector of strings.
+ *  1)` Caller expected to call init first, which initiates the connection
+ *      to the capture end point. Being async, it would take some milliseconds
+ *      to connect.
  *
- * Another way, the entire cache in either direction can be sent/received
- * via PAIR socket. But this woulkd need a special terminating message to 
- * indicate read end, as non-blocking read returning no event does not 
- * necessarily mean end
+ *  2)  Caller starts the cache, optionally with some local cache it may have.
+ *      The cache service keeps it as its startup/initial stock.
+ *      This helps the caller saves his local cache with cache service.
  *
- * Not sure, what the gain here is vs current approach of vector<string>
- * Just a note for now, not to lose a line of possibility.
+ *  3)  Caller call stops, upon it making connect to XPUB end.
+ *      As caller's connect is async and also this zmq end may have some cache
+ *      of events by ZMQ locally. So read events little longer.
+ *
+ *  4)  Upon stop, the caller may read cached events.
+ *      The events are provided in FIFO order. 
+ *      As cached events can be too many, the service returns a few at a time.
+ *      The caller is expected to read repeatedly until no event is returned.
+ *
+ *  Cache overflow:
+ *      A ceil is set and may run out of memory, before ceil is reached.
+ *      In either case, the caching is *not* completely stopped but cached as
+ *      one event per runtime-id/publishing instance. This info is required
+ *      to compute missed message count due to overflow and otherwise.
  */
 
 int