forb: Set cancelation points in receiver thread

[frescor/forb.git] / src / iop.c

diff --git a/src/iop.c b/src/iop.c
index 51330693aba314314b387304cfc6f198b840d9a2..5a937aa344137a9afcb3f71e7c39a4325692e814 100644 (file)
--- a/src/iop.c
+++ b/src/iop.c
@@ -91,17 +91,15 @@ forb_iop_prepend_message_header(FORB_CDR_Codec *codec, forb_iop_message_type mt)
 
 CORBA_boolean
 forb_iop_prepare_request(forb_request_t *req,
-                        char *iface,
-                        unsigned method_ind,
                         CORBA_Environment *env)
 {
        CORBA_boolean ret;
        forb_iop_request_header rh;
 
        rh.request_id = req->request_id;
-       rh.iface = iface;
+       rh.iface = req->interface;
        rh.objkey = forb_object_to_key(req->obj);
-       rh.method_index = method_ind;
+       rh.method_index = req->method_ind;
        rh.source = forb_object_to_forb(req->obj)->server_id;
        ret = forb_iop_request_header_serialize(&req->cdr_request, &rh);
        if (ret) {
@@ -110,8 +108,9 @@ forb_iop_prepare_request(forb_request_t *req,
                char str[50];
                ul_logdeb("preparing request: id=%d  dest=%s  iface=%s method=%d\n", req->request_id,
                          forb_server_id_to_string(str, &req->obj->server, sizeof(str)),
-                         iface, method_ind);
+                         rh.iface, rh.method_index);
        }
+       req->end_of_header_index = req->cdr_request.wptr;
        return ret;
 }
 
@@ -388,6 +387,7 @@ process_request(forb_port_t *port, FORB_CDR_Codec *codec, uint32_t message_size)
        exec_req = forb_malloc(sizeof(*exec_req));
        if (exec_req) {
                memset(exec_req, 0, sizeof(exec_req));
+               exec_req->request_type = FORB_EXEC_REQ_REMOTE; 
                exec_req->request_id = request_header.request_id;
                exec_req->source = request_header.source;
                exec_req->obj = obj;
@@ -597,12 +597,15 @@ void *forb_iop_receiver_thread(void *arg)
        size_t len;
        forb_iop_message_header mh;
        bool header_received = false;
+       int oldstate;
 
        while (!port->finish) {
                if (c->rptr == c->wptr) {
                        /* The buffer is empty now - start writing from begining*/
                        FORB_CDR_buffer_reset(c, 0);
                }
+
+               pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &oldstate);
                /* TODO: If there is not enough space for reception,
                 * we should shift the already received data to the
                 * beginning of the buffer. */
@@ -616,10 +619,12 @@ void *forb_iop_receiver_thread(void *arg)
                c->wptr += rcvd;
                 c->wptr_last = c->wptr;
 
+               pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &oldstate);
+
                /* While there are some data in the buffer, process them. */
                while (FORB_CDR_data_size(c) > 0) {
                        len = FORB_CDR_data_size(c);
-                       /* Wait for and then process message header */
+                       /* Find and process message header (if there is any) */
                        if (!header_received) {
                                if (len >= forb_iop_MESSAGE_HEADER_SIZE) {
                                        if (c->rptr % 8 != 0) {
@@ -636,7 +641,7 @@ void *forb_iop_receiver_thread(void *arg)
                                }
                        }
                        
-                       /* Wait for and then process the message body */
+                       /* If the whole message is received, then process it */
                        if (header_received) {
                                if (len >= mh.message_size) {
                                        process_message(port, &mh, c);
@@ -711,11 +716,13 @@ void *forb_iop_discovery_thread(void *arg)
 }
 
 /** 
- * Sends REQUEST message to another FORB.
+ * Sends REQUEST to another object.
  *
- * The request @a req has to be prepared by
- * forb_iop_prepare_request(). Then, this function adds a message
- * header, connects to the destination FORB and sends the request.
+ * The request @a req has to be prepared previously by calling
+ * forb_iop_prepare_request(). Then, when the request destination is
+ * remote, this function adds a message header, connects to the
+ * destination FORB and sends the request. In the case of local
+ * request, it is directly enqueued into the executor's queue.
  *
  * If no exception is reported, then the caller must wait for response
  * by calling forb_wait_for_reply().
@@ -732,6 +739,7 @@ forb_request_send(forb_request_t *req, CORBA_Environment *env)
        size_t len;
        fosa_abs_time_t timeout;
        forb_t *forb = forb_object_to_forb(req->obj);
+       forb_exec_req_t *exec_req;
 
        if (!forb) {
                env->major = FORB_EX_INTERNAL;
@@ -748,6 +756,24 @@ forb_request_send(forb_request_t *req, CORBA_Environment *env)
                return;
        }
 
+       /* Local invocation case, destination of a message is only 
+        * a different executor thread */
+       if (forb_object_is_local(req->obj)) {
+               exec_req = forb_malloc(sizeof(*exec_req));
+               memset(exec_req, 0, sizeof(exec_req));
+               exec_req->request_type = FORB_EXEC_REQ_LOCAL; 
+               exec_req->input_request = req;
+               exec_req->obj = forb_object_duplicate(req->obj);
+               exec_req->method_index = req->method_ind;
+               exec_req->interface = req->interface;
+               req->cdr_request.rptr = req->end_of_header_index;
+               exec_req->codec = req->cdr_request;
+               req->cdr_request.release_buffer = CORBA_FALSE;
+               exec_req->request_id = req->request_id;
+               forb_exec_req_ins_tail(forb_object_get_executor(exec_req->obj), exec_req);
+               return;
+       }
+
        ret = forb_iop_prepend_message_header(&req->cdr_request, forb_iop_REQUEST);
        if (!ret) {
                /* This should never happen */