add infrastructure to cleanly swap inbetween the DB and the IO queue; add infrastruct...

  - QueueDBOperation(): use our own vars on the AsyncIO struct, so we don't overwrite handlers & events from the file I/O stuff.
  - DB_PerformNext(): don't use default: so we get warnings if new states are added and not handled here.
  - DB_PerformNext(): if we don't get a DB query or Terminate, the db operations are done, remove our handlers, this AsyncIO context is handed over into another Queue by the application logic.
  - NextDBOperation(): use our own vars on the AsyncIO struct, so we don't overwrite handlers & events from the file I/O stuff.
  - HandleInbound(): handle more read-states; needed for message reading.
  - HandleInbound(): move Rumpelstilskin lookup if to switch()
  - IO_send_callback(): fix/add IO debugging to /tmp/foo
  - IO_send_callback(): Adjust switch to some edge cases:
    - eSendReply: check whether we're done with sending, if, fall forward to writing.
    - eReadMore: was missing, continue reading (if)
    - eReadPayload: new, read smtp messages for example.
    - eDBQuery: the application logic want to perform database I/O with this context, clean up hooks and remove it from this queue.
  - set_start_callback(): properly hanle the new cases
  - IO_recv_callback(): fix/add IO debugging to /tmp/foo
  - ReAttachIO(): use me if you want to continue with file I/O after i.e. DB-I/O

diff --git a/citadel/event_client.c b/citadel/event_client.c
index 68e957a8ed7401005cc09c27eadae7031459d5a5..68e59cc1f11a36da456117a8d2b506b70e4a4f43 100644 (file)
--- a/citadel/event_client.c
+++ b/citadel/event_client.c
@@ -95,10 +95,10 @@ int QueueDBOperation(AsyncIO *IO, IO_CallBack CB)
        h = (IOAddHandler*)malloc(sizeof(IOAddHandler));
        h->IO = IO;
        h->EvAttch = CB;
-       ev_cleanup_init(&IO->abort_by_shutdown, 
+       ev_cleanup_init(&IO->db_abort_by_shutdown, 
                        IO_abort_shutdown_callback);
-       IO->abort_by_shutdown.data = IO;
-       ev_cleanup_start(event_db, &IO->abort_by_shutdown);
+       IO->db_abort_by_shutdown.data = IO;
+       ev_cleanup_start(event_db, &IO->db_abort_by_shutdown);
 
        citthread_mutex_lock(&DBEventQueueMutex);
        CtdlLogPrintf(CTDL_DEBUG, "DBEVENT Q\n");
@@ -117,7 +117,7 @@ void ShutDownDBCLient(AsyncIO *IO)
        become_session(Ctx);
 
        CtdlLogPrintf(CTDL_DEBUG, "DBEVENT\n");
-       ev_cleanup_stop(event_db, &IO->abort_by_shutdown);
+       ev_cleanup_stop(event_db, &IO->db_abort_by_shutdown);
 
        assert(IO->Terminate);
        IO->Terminate(IO);      
@@ -133,25 +133,37 @@ DB_PerformNext(struct ev_loop *loop, ev_idle *watcher, int revents)
        CtdlLogPrintf(CTDL_DEBUG, "event: %s\n", __FUNCTION__);
        become_session(IO->CitContext);
        
-       ev_idle_stop(event_db, &IO->unwind_stack);
+       ev_idle_stop(event_db, &IO->db_unwind_stack);
 
-       assert(IO->ReadDone);
-       switch (IO->ReadDone(IO))
+       assert(IO->NextDBOperation);
+       switch (IO->NextDBOperation(IO))
        {
+       case eDBQuery:
+               break;
+       case eSendDNSQuery:
+       case eReadDNSReply:
+       case eConnect:
+       case eSendReply: 
+       case eSendMore:
+       case eReadMessage: 
+       case eReadMore:
+       case eReadPayload:
+               ev_cleanup_stop(loop, &IO->db_abort_by_shutdown);
+               break;
+       case eTerminateConnection:
        case eAbort:
            ShutDownDBCLient(IO);
-       default:
-           break;
        }
 }
 
-void NextDBOperation(AsyncIO *IO, IO_CallBack CB)
+eNextState NextDBOperation(AsyncIO *IO, IO_CallBack CB)
 {
-       IO->ReadDone = CB;
-       ev_idle_init(&IO->unwind_stack,
+       IO->NextDBOperation = CB;
+       ev_idle_init(&IO->db_unwind_stack,
                     DB_PerformNext);
-       IO->unwind_stack.data = IO;
-       ev_idle_start(event_db, &IO->unwind_stack);
+       IO->db_unwind_stack.data = IO;
+       ev_idle_start(event_db, &IO->db_unwind_stack);
+       return eDBQuery;
 }
 
 /*--------------------------------------------------------------------------------
@@ -255,7 +267,11 @@ eReadState HandleInbound(AsyncIO *IO)
        
        become_session(IO->CitContext);
 
-       while ((Finished == eBufferNotEmpty) && (IO->NextState == eReadMessage)){
+       while ((Finished == eBufferNotEmpty) && 
+              ((IO->NextState == eReadMessage)||
+               (IO->NextState == eReadMore)||
+               (IO->NextState == eReadPayload)))
+       {
                if (IO->RecvBuf.nBlobBytesWanted != 0) { 
                                
                }
@@ -288,17 +304,28 @@ eReadState HandleInbound(AsyncIO *IO)
                }
        }
 
-
-       if ((IO->NextState == eSendReply) ||
-           (IO->NextState == eSendMore))
-       {
+       switch (IO->NextState) {
+       case eSendReply:
+       case eSendMore:
                assert(IO->SendDone);
                IO->NextState = IO->SendDone(IO);
                ev_io_start(event_base, &IO->send_event);
-       }
-       else if ((IO->NextState == eTerminateConnection) ||
-                (IO->NextState == eAbort) )
+               break;
+       case eReadPayload:
+       case eReadMore:
+               ev_io_start(event_base, &IO->recv_event);
+               break;
+       case eTerminateConnection:
+       case eAbort:
                ShutDownCLient(IO);
+               break;
+       case eSendDNSQuery:
+       case eReadDNSReply:
+       case eDBQuery:
+       case eConnect:
+       case eReadMessage:
+               break;
+       }
        return Finished;
 }
 
@@ -310,6 +337,34 @@ IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents)
        AsyncIO *IO = watcher->data;
 
        become_session(IO->CitContext);
+#ifdef BIGBAD_IODBG
+       {
+               int rv = 0;
+               char fn [SIZ];
+               FILE *fd;
+               const char *pch = ChrPtr(IO->SendBuf.Buf);
+               const char *pchh = IO->SendBuf.ReadWritePointer;
+               long nbytes;
+               
+               if (pchh == NULL)
+                       pchh = pch;
+               
+               nbytes = StrLength(IO->SendBuf.Buf) - (pchh - pch);
+               snprintf(fn, SIZ, "/tmp/foolog_ev_%s.%d",
+                        ((CitContext*)(IO->CitContext))->ServiceName,
+                        IO->SendBuf.fd);
+               
+               fd = fopen(fn, "a+");
+               fprintf(fd, "Read: BufSize: %ld BufContent: [",
+                       nbytes);
+               rv = fwrite(pchh, nbytes, 1, fd);
+               if (!rv) printf("failed to write debug to %s!\n", fn);
+               fprintf(fd, "]\n");
+               
+               
+               fclose(fd);
+       }
+#endif
        rc = StrBuf_write_one_chunk_callback(watcher->fd, 0/*TODO*/, &IO->SendBuf);
 
        if (rc == 0)
@@ -327,13 +382,15 @@ IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents)
                                pchh = pch;
                        
                        nbytes = StrLength(IO->SendBuf.Buf) - (pchh - pch);
-                       snprintf(fn, SIZ, "/tmp/foolog_ev_%s.%d", "smtpev", IO->SendBuf.fd);
+                       snprintf(fn, SIZ, "/tmp/foolog_ev_%s.%d", 
+                                ((CitContext*)(IO->CitContext))->ServiceName, 
+                                IO->SendBuf.fd);
                
                        fd = fopen(fn, "a+");
                        fprintf(fd, "Read: BufSize: %ld BufContent: [",
                                nbytes);
                        rv = fwrite(pchh, nbytes, 1, fd);
-                       if (!rv) printf("failed to write debug!");
+                       if (!rv) printf("failed to write debug to %s!\n", fn);
                        fprintf(fd, "]\n");
                
                        
@@ -342,8 +399,6 @@ IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents)
 #endif
                ev_io_stop(event_base, &IO->send_event);
                switch (IO->NextState) {
-               case eSendReply:
-                       break;
                case eSendMore:
                        assert(IO->SendDone);
                        IO->NextState = IO->SendDone(IO);
@@ -355,7 +410,13 @@ IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents)
                                ev_io_start(event_base, &IO->send_event);
                        }
                        break;
+               case eSendReply:
+                   if (StrBufCheckBuffer(&IO->SendBuf) != eReadSuccess) 
+                       break;
+                   IO->NextState = eReadMore;
+               case eReadMore:
                case eReadMessage:
+               case eReadPayload:
                        if (StrBufCheckBuffer(&IO->RecvBuf) == eBufferNotEmpty) {
                                HandleInbound(IO);
                        }
@@ -363,6 +424,10 @@ IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents)
                                ev_io_start(event_base, &IO->recv_event);
                        }
 
+                       break;
+               case eDBQuery:
+                       /* we now live in another queue, so we have to unregister. */
+                       ev_cleanup_stop(loop, &IO->abort_by_shutdown);
                        break;
                case eSendDNSQuery:
                case eReadDNSReply:
@@ -383,14 +448,17 @@ set_start_callback(struct ev_loop *loop, AsyncIO *IO, int revents)
 {
        
        switch(IO->NextState) {
+       case eReadMore:
        case eReadMessage:
                ev_io_start(event_base, &IO->recv_event);
                break;
        case eSendReply:
        case eSendMore:
+       case eReadPayload:
                become_session(IO->CitContext);
                IO_send_callback(loop, &IO->send_event, revents);
                break;
+       case eDBQuery:
        case eSendDNSQuery:
        case eReadDNSReply:
        case eConnect:
@@ -498,6 +566,34 @@ IO_recv_callback(struct ev_loop *loop, ev_io *watcher, int revents)
        AsyncIO *IO = watcher->data;
 
        nbytes = StrBuf_read_one_chunk_callback(watcher->fd, 0 /*TODO */, &IO->RecvBuf);
+#ifdef BIGBAD_IODBG
+       {
+               int rv = 0;
+               char fn [SIZ];
+               FILE *fd;
+               const char *pch = ChrPtr(IO->RecvBuf.Buf);
+               const char *pchh = IO->RecvBuf.ReadWritePointer;
+               long nbytes;
+               
+               if (pchh == NULL)
+                       pchh = pch;
+               
+               nbytes = StrLength(IO->RecvBuf.Buf) - (pchh - pch);
+               snprintf(fn, SIZ, "/tmp/foolog_ev_%s.%d",
+                        ((CitContext*)(IO->CitContext))->ServiceName, 
+                        IO->SendBuf.fd);
+               
+               fd = fopen(fn, "a+");
+               fprintf(fd, "Read: BufSize: %ld BufContent: [",
+                       nbytes);
+               rv = fwrite(pchh, nbytes, 1, fd);
+               if (!rv) printf("failed to write debug to %s!\n", fn);
+               fprintf(fd, "]\n");
+               
+               
+               fclose(fd);
+       }
+#endif
        if (nbytes > 0) {
                HandleInbound(IO);
        } else if (nbytes == 0) {
@@ -643,3 +739,21 @@ eNextState InitEventIO(AsyncIO *IO,
        }
        return event_connect_socket(IO, conn_timeout, first_rw_timeout);
 }
+
+eNextState ReAttachIO(AsyncIO *IO, 
+                     void *pData, 
+                     int ReadFirst)
+{
+       IO->Data = pData;
+       become_session(IO->CitContext);
+       ev_cleanup_start(event_base, &IO->abort_by_shutdown);
+       if (ReadFirst) {
+               IO->NextState = eReadMessage;
+       }
+       else {
+               IO->NextState = eSendReply;
+       }
+       set_start_callback(event_base, IO, 0);
+
+       return IO->NextState;
+}

diff --git a/citadel/event_client.h b/citadel/event_client.h
index 9ed88c01e15401a16d00ed26242d34c08eb3653f..99739e0332d26ffc731988070bbcbc57c367477f 100644 (file)
--- a/citadel/event_client.h
+++ b/citadel/event_client.h
@@ -14,11 +14,13 @@ typedef struct AsyncIO AsyncIO;
 typedef enum _eNextState {
        eSendDNSQuery,
        eReadDNSReply,
+       eDBQuery,
        eConnect,
        eSendReply, 
        eSendMore,
        eReadMessage, 
        eReadMore,
+       eReadPayload,
        eTerminateConnection,
        eAbort
 }eNextState;
@@ -28,6 +30,21 @@ typedef eReadState (*IO_LineReaderCallback)(AsyncIO *IO);
 typedef void (*ParseDNSAnswerCb)(AsyncIO*, unsigned char*, int);
 typedef void (*FreeDNSReply)(void *DNSData);
 
+
+typedef struct __ReadAsyncMsg {
+       StrBuf *MsgBuf;
+       size_t maxlen;          /* maximum message length */
+
+       const char *terminator; /* token signalling EOT */
+       long tlen;
+       int dodot;
+
+       int flushing;           /* if we read maxlen, read until nothing more arives and ignore this. */
+
+       int crlf;               /* CRLF newlines instead of LF */
+} ReadAsyncMsg;
+
+
 typedef struct _DNSQueryParts {
        ParseDNSAnswerCb DNS_CB;
        IO_CallBack PostDNS;
@@ -66,10 +83,12 @@ struct AsyncIO {
                RecvBuf;
 
        /* our events... */
-       ev_cleanup abort_by_shutdown; /* server wants to go down... */
+       ev_cleanup abort_by_shutdown, /* server wants to go down... */
+               db_abort_by_shutdown; /* server wants to go down... */
        ev_timer conn_fail,           /* connection establishing timed out */
                rw_timeout;           /* timeout while sending data */
        ev_idle unwind_stack,         /* get c-ares out of the stack */
+               db_unwind_stack,      /* wait for next db operation... */
                conn_fail_immediate;  /* unwind stack, but fail immediately. */
        ev_io recv_event,             /* receive data from the client */
                send_event,           /* send more data to the client */
@@ -83,7 +102,8 @@ struct AsyncIO {
                Terminate,    /* shutting down... */
                Timeout,      /* Timeout handler; may also be connection timeout */
                ConnFail,     /* What to do when one connection failed? */
-               ShutdownAbort;/* we're going down. make your piece. */ 
+               ShutdownAbort,/* we're going down. make your piece. */ 
+               NextDBOperation; /* Perform Database IO */
 
        IO_LineReaderCallback LineReader; /* if we have linereaders, maybe we want to read more lines before the real application logic is called? */
 
@@ -97,7 +117,7 @@ struct AsyncIO {
        evcurl_request_data HttpReq;
 
        /* Saving / loading a message async from / to disk */
-
+       ReadAsyncMsg *ReadMsg;
        struct CtdlMessage *AsyncMsg;
        struct recptypes *AsyncRcp;
        /* Custom data; its expected to contain  AsyncIO so we can save malloc()s... */
@@ -112,7 +132,7 @@ typedef struct _IOAddHandler {
 
 void FreeAsyncIOContents(AsyncIO *IO);
 
-void NextDBOperation(AsyncIO *IO, IO_CallBack CB);
+eNextState NextDBOperation(AsyncIO *IO, IO_CallBack CB);
 int QueueDBOperation(AsyncIO *IO, IO_CallBack CB);
 int QueueEventContext(AsyncIO *IO, IO_CallBack CB);
 int ShutDownEventQueue(void);
@@ -153,6 +173,10 @@ int evcurl_init(AsyncIO *IO,
                IO_CallBack CallBack, 
                IO_CallBack Terminate);
 
+eNextState ReAttachIO(AsyncIO *IO, 
+                     void *pData, 
+                     int ReadFirst);
+
 void evcurl_handle_start(AsyncIO *IO);
 
 #endif /* __EVENT_CLIENT_H__ */