channel_*

lib/chansx.c

@@ -100,3 +100,152 @@ uint8_t _channel_close(conn_t *co, uint16_t chid)
 
   return SNE_SUCCESS;
 }
+
+chnl_t *sxchannel_open(conn_t *co, int type)
+{
+  chnl_t *chan = NULL;
+  sxmsg_t *msg = NULL;
+  sntllv2_head_t *head;
+  int msgidx, r;
+
+  if(!co) {
+    r = SNE_FAILED;
+    goto __reterr;
+  }
+
+  if(!(chan = malloc(sizeof(chnl_t)))) {
+  __enomem:
+    r = SNE_ENOMEM;
+    goto __reterr;
+  }
+  if(!(msg = malloc(sizeof(sxmsg_t)))) goto __enomem;
+
+  /* early init for the channel */
+  chan->connection = co;
+  chan->flags = 0;
+
+  /* early init for the message */
+  pthread_mutex_init(&msg->wait, NULL);
+  pthread_mutex_lock(&msg->wait);
+  msg->pch = chan;
+  msg->payload = NULL;
+  memset(&msg->mhead, 0, sizeof(sntllv2_head_t));
+  head = &msg->mhead;
+
+  /* form a header */
+  head->attr = SXMSG_PROTO | SXMSG_REPLYREQ | SXMSG_OPEN;
+  head->payload_length = 0;
+  head->reserve = (uint16_t)type;
+
+  /* try to alloc a message */
+  pthread_mutex_lock(&co->idx_msg_lock);
+  msgidx = idx_allocate(&co->idx_msg);
+  if(msgidx != IDX_INVAL) co->messages[msgidx] = msg;
+  pthread_mutex_unlock(&co->idx_msg_lock);
+
+  if(msgidx == IDX_INVAL) {    r = SNE_MMESSAGES;    goto __reterr2;  }
+
+  /* now we're ready to write it */
+  r = _sntll_writemsg(co, msg);
+  if(r == SNE_SUCCESS) pthread_mutex_lock(&msg->wait);
+  else goto __reterr3;
+
+  /* we will wakeup on return */
+  if(msg->mhead.opcode != SNE_SUCCESS) { r = msg->mhead.opcode; goto __reterr3;  }
+
+  /* ok all is fine */
+  chan->cid = msg->mhead.reserve;
+  pthread_mutex_lock(&co->idx_ch_lock);
+  idx_reserve(&co->idx_ch, msg->mhead.reserve);
+  co->channels[msg->mhead.reserve] = chan;
+  pthread_mutex_unlock(&co->idx_ch_lock);
+
+  /* destroy a message */
+  pthread_mutex_lock(&co->idx_msg_lock);
+  idx_free(&co->idx_msg, msgidx);
+  co->messages[msgidx] = NULL;
+  pthread_mutex_unlock(&co->idx_msg_lock);
+
+  /* free allocated resources */
+  pthread_mutex_unlock(&msg->wait);
+  pthread_mutex_destroy(&msg->wait);
+  free(msg);
+
+  return chan;
+
+ __reterr3:
+  pthread_mutex_lock(&co->idx_msg_lock);
+  idx_free(&co->idx_msg, msgidx);
+  co->messages[msgidx] = NULL;
+  pthread_mutex_unlock(&co->idx_msg_lock);
+ __reterr2:
+  pthread_mutex_unlock(&msg->wait);
+  pthread_mutex_destroy(&msg->wait);
+ __reterr:
+  if(chan) free(chan);
+  if(msg) free(msg);
+  errno = r;
+  return NULL;
+}
+
+int sxchannel_close(chnl_t *channel)
+{
+  int r = SNE_SUCCESS;
+  sxmsg_t *msg;
+  sntllv2_head_t *head;
+  conn_t *co;
+  int msgidx;
+
+  /* check channel validity */
+  if(!channel) return SNE_FAILED;
+  else if(!(co = channel->connection)) return SNE_FAILED;
+  if(channel->cid > 512) return SNE_IGNORED;
+  if(channel != co->channels[channel->cid]) return SNE_IGNORED;
+
+  if(!(msg = malloc(sizeof(sxmsg_t)))) return SNE_ENOMEM;
+  head = &msg->mhead;
+  memset(head, 0, sizeof(sntllv2_head_t));
+
+  /* setup head */
+  head->attr = SXMSG_PROTO | SXMSG_REPLYREQ; /* close channel */
+  head->reserve = channel->cid;
+
+  /* setup message */
+  pthread_mutex_init(&msg->wait, NULL);
+  pthread_mutex_lock(&msg->wait);
+  msg->pch = channel;
+
+  /* allocate it */
+  pthread_mutex_lock(&co->idx_msg_lock);
+  msgidx = idx_allocate(&co->idx_msg);
+  if(msgidx != IDX_INVAL) co->messages[msgidx] = msg;
+  pthread_mutex_unlock(&co->idx_msg_lock);
+
+  if(msgidx == IDX_INVAL) {    r = SNE_MMESSAGES;    goto __reterr2;  }
+
+  r = _sntll_writemsg(co, msg);
+  if(r == SNE_SUCCESS) {
+    pthread_mutex_lock(&msg->wait);
+    r = head->opcode;
+
+    /* we will free this anyway */
+    pthread_mutex_lock(&co->idx_ch_lock);
+    idx_free(&co->idx_ch, channel->cid);
+    co->channels[channel->cid] = NULL;
+    pthread_mutex_unlock(&co->idx_ch_lock);
+    free(channel);
+  }
+
+  pthread_mutex_lock(&co->idx_msg_lock);
+  idx_free(&co->idx_msg, msgidx);
+  co->messages[msgidx] = NULL;
+  pthread_mutex_unlock(&co->idx_msg_lock);
+
+ __reterr2:
+  pthread_mutex_unlock(&msg->wait);
+  pthread_mutex_destroy(&msg->wait);
+  free(msg);
+
+  return r;
+}
+