summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/lib/cdap.c166
-rw-r--r--src/lib/cdap_req.c6
-rw-r--r--src/lib/cdap_req.h8
3 files changed, 114 insertions, 66 deletions
diff --git a/src/lib/cdap.c b/src/lib/cdap.c
index a60bf8f9..f5e0593d 100644
--- a/src/lib/cdap.c
+++ b/src/lib/cdap.c
@@ -39,12 +39,12 @@
#include "cdap.pb-c.h"
typedef Cdap cdap_t;
-typedef int32_t invoke_id_t;
+typedef cdap_key_t invoke_id_t;
#define CDAP_REPLY (CDAP_DELETE + 1)
-#define INVALID_INVOKE_ID -1
-#define IDS_SIZE 256
+#define INVALID_ID -1
+#define IDS_SIZE 2048
#define BUF_SIZE 2048
struct fd_el {
@@ -78,8 +78,10 @@ struct cdap_rcvd {
struct list_head next;
int fd;
+ bool proc;
invoke_id_t iid;
+ cdap_key_t key;
enum cdap_opcode opcode;
char * name;
@@ -88,7 +90,7 @@ struct cdap_rcvd {
uint32_t flags;
};
-static int next_invoke_id(struct cdap * instance)
+static int next_id(struct cdap * instance)
{
int ret;
@@ -98,15 +100,15 @@ static int next_invoke_id(struct cdap * instance)
ret = bmp_allocate(instance->ids);
if (!bmp_is_id_valid(instance->ids, ret))
- ret = INVALID_INVOKE_ID;
+ ret = INVALID_ID;
pthread_mutex_unlock(&instance->ids_lock);
return ret;
}
-static int release_invoke_id(struct cdap * instance,
- int id)
+static int release_id(struct cdap * instance,
+ int32_t id)
{
int ret;
@@ -130,7 +132,6 @@ static struct cdap_req * cdap_sent_get_by_key(struct cdap * instance,
struct cdap_req * req = NULL;
assert(instance);
- assert(key >= 0);
pthread_rwlock_rdlock(&instance->sent_lock);
@@ -147,21 +148,43 @@ static struct cdap_req * cdap_sent_get_by_key(struct cdap * instance,
return NULL;
}
-static struct cdap_rcvd * cdap_rcvd_get_by_iid(struct cdap * instance,
- invoke_id_t iid)
+static struct cdap_req * cdap_sent_get_by_iid(struct cdap * instance,
+ invoke_id_t iid)
+{
+ struct list_head * p = NULL;
+ struct cdap_req * req = NULL;
+
+ assert(instance);
+
+ pthread_rwlock_rdlock(&instance->sent_lock);
+
+ list_for_each(p, &instance->sent) {
+ req = list_entry(p, struct cdap_req, next);
+ if (req->iid == iid) {
+ pthread_rwlock_unlock(&instance->sent_lock);
+ return req;
+ }
+ }
+
+ pthread_rwlock_unlock(&instance->sent_lock);
+
+ return NULL;
+}
+
+static struct cdap_rcvd * cdap_rcvd_get_by_key(struct cdap * instance,
+ cdap_key_t key)
{
struct list_head * p = NULL;
struct list_head * h = NULL;
struct cdap_rcvd * rcvd = NULL;
assert(instance);
- assert(iid >= 0);
pthread_mutex_lock(&instance->rcvd_lock);
list_for_each_safe(p, h, &instance->rcvd) {
rcvd = list_entry(p, struct cdap_rcvd, next);
- if (rcvd->iid == iid) {
+ if (rcvd->key == key) {
pthread_mutex_unlock(&instance->rcvd_lock);
list_del(&rcvd->next);
return rcvd;
@@ -170,20 +193,22 @@ static struct cdap_rcvd * cdap_rcvd_get_by_iid(struct cdap * instance,
pthread_mutex_unlock(&instance->rcvd_lock);
+ assert(false);
+
return NULL;
}
static struct cdap_req * cdap_sent_add(struct cdap * instance,
int fd,
+ invoke_id_t iid,
cdap_key_t key)
{
struct cdap_req * req;
assert(instance);
- assert(key >= 0);
assert(!cdap_sent_has_key(instance, key));
- req = cdap_req_create(fd, key);
+ req = cdap_req_create(fd, iid, key);
if (req == NULL)
return NULL;
@@ -285,11 +310,20 @@ static void * sdu_reader(void * o)
assert(msg->name);
rcvd->opcode = msg->opcode;
+ rcvd->fd = fd;
rcvd->iid = msg->invoke_id;
+ rcvd->key = next_id(instance);
+ if (rcvd->key == INVALID_ID) {
+ cdap__free_unpacked(msg, NULL);
+ free(rcvd);
+ continue;
+ }
+
rcvd->flags = msg->flags;
- rcvd->fd = fd;
+ rcvd->proc = false;
rcvd->name = strdup(msg->name);
if (rcvd->name == NULL) {
+ release_id(instance, rcvd->key);
cdap__free_unpacked(msg, NULL);
free(rcvd);
continue;
@@ -299,6 +333,7 @@ static void * sdu_reader(void * o)
rcvd->len = msg->value.len;
rcvd->data = malloc(rcvd->len);
if (rcvd->data == NULL) {
+ release_id(instance, rcvd->key);
cdap__free_unpacked(msg, NULL);
free(rcvd->name);
free(rcvd);
@@ -317,7 +352,7 @@ static void * sdu_reader(void * o)
pthread_cond_signal(&instance->rcvd_cond);
pthread_mutex_unlock(&instance->rcvd_lock);
} else {
- req = cdap_sent_get_by_key(instance, msg->invoke_id);
+ req = cdap_sent_get_by_iid(instance, msg->invoke_id);
if (req == NULL) {
cdap__free_unpacked(msg, NULL);
continue;
@@ -569,22 +604,6 @@ static int write_msg(int fd,
return 0;
}
-static cdap_key_t invoke_id_to_key(invoke_id_t iid)
-{
- if (iid == INVALID_INVOKE_ID)
- return INVALID_CDAP_KEY;
-
- return (cdap_key_t) iid;
-}
-
-static invoke_id_t key_to_invoke_id(cdap_key_t key)
-{
- if (key == INVALID_CDAP_KEY)
- return INVALID_INVOKE_ID;
-
- return (invoke_id_t) key;
-}
-
cdap_key_t * cdap_request_send(struct cdap * instance,
enum cdap_opcode code,
const char * name,
@@ -629,22 +648,28 @@ cdap_key_t * cdap_request_send(struct cdap * instance,
invoke_id_t iid;
struct fd_el * e;
- iid = next_invoke_id(instance);
- if (iid == INVALID_INVOKE_ID) {
+ iid = next_id(instance);
+ if (iid == INVALID_ID) {
pthread_rwlock_unlock(&instance->flows_lock);
return keys;
}
msg.invoke_id = iid;
- *key = invoke_id_to_key(iid);
-
e = list_entry(p, struct fd_el, next);
- req = cdap_sent_add(instance, e->fd, *key);
+ *key = next_id(instance);
+ if (*key == INVALID_ID) {
+ pthread_rwlock_unlock(&instance->flows_lock);
+ release_id(instance, iid);
+ return keys;
+ }
+
+ req = cdap_sent_add(instance, e->fd, iid, *key);
if (req == NULL) {
pthread_rwlock_unlock(&instance->flows_lock);
- release_invoke_id(instance, iid);
+ release_id(instance, *key);
+ release_id(instance, iid);
return keys;
}
@@ -652,13 +677,15 @@ cdap_key_t * cdap_request_send(struct cdap * instance,
if (ret == -ENOMEM) {
pthread_rwlock_unlock(&instance->flows_lock);
cdap_sent_del(instance, req);
- release_invoke_id(instance, iid);
+ release_id(instance, *key);
+ release_id(instance, iid);
return keys;
}
if (ret < 0) {
cdap_sent_del(instance, req);
- release_invoke_id(instance, iid);
+ release_id(instance, *key);
+ release_id(instance, iid);
}
++key;
@@ -676,10 +703,8 @@ int cdap_reply_wait(struct cdap * instance,
{
int ret;
struct cdap_req * r;
- invoke_id_t iid = key_to_invoke_id(key);
- if (instance == NULL || iid == INVALID_INVOKE_ID
- || (data != NULL && len == NULL))
+ if (instance == NULL || (data != NULL && len == NULL))
return -EINVAL;
r = cdap_sent_get_by_key(instance, key);
@@ -689,7 +714,7 @@ int cdap_reply_wait(struct cdap * instance,
ret = cdap_req_wait(r);
if (ret < 0) {
cdap_sent_del(instance, r);
- release_invoke_id(instance, iid);
+ release_id(instance, r->iid);
return ret;
}
@@ -703,7 +728,7 @@ int cdap_reply_wait(struct cdap * instance,
ret = r->response;
cdap_sent_del(instance, r);
- release_invoke_id(instance, iid);
+ release_id(instance, r->iid);
return ret;
}
@@ -715,7 +740,7 @@ cdap_key_t cdap_request_wait(struct cdap * instance,
size_t * len,
uint32_t * flags)
{
- struct cdap_rcvd * rcvd;
+ struct cdap_rcvd * rcv = NULL;
if (instance == NULL || opcode == NULL || name == NULL || data == NULL
|| len == NULL || flags == NULL)
@@ -726,23 +751,35 @@ cdap_key_t cdap_request_wait(struct cdap * instance,
pthread_cleanup_push((void(*)(void *))pthread_mutex_unlock,
(void *) &instance->rcvd_lock);
- while (list_is_empty(&instance->rcvd))
- pthread_cond_wait(&instance->rcvd_cond, &instance->rcvd_lock);
+ while (rcv == NULL) {
+ while (list_is_empty(&instance->rcvd))
+ pthread_cond_wait(&instance->rcvd_cond,
+ &instance->rcvd_lock);
- rcvd = list_first_entry(&instance->rcvd, struct cdap_rcvd, next);
+ rcv = list_first_entry(&instance->rcvd, struct cdap_rcvd, next);
+ if (rcv->proc) {
+ rcv = NULL;
+ pthread_cond_wait(&instance->rcvd_cond,
+ &instance->rcvd_lock);
+ }
+ }
+
+ rcv->proc = true;
+ list_del(&rcv->next);
+ list_add_tail(&rcv->next, &instance->rcvd);
pthread_cleanup_pop(true);
- *opcode = rcvd->opcode;
- *name = rcvd->name;
- *data = rcvd->data;
- *len = rcvd->len;
- *flags = rcvd->flags;
+ *opcode = rcv->opcode;
+ *name = rcv->name;
+ *data = rcv->data;
+ *len = rcv->len;
+ *flags = rcv->flags;
- rcvd->name = NULL;
- rcvd->data = NULL;
+ rcv->name = NULL;
+ rcv->data = NULL;
- return invoke_id_to_key(rcvd->iid);
+ return rcv->key;
}
int cdap_reply_send(struct cdap * instance,
@@ -753,16 +790,17 @@ int cdap_reply_send(struct cdap * instance,
{
int fd;
cdap_t msg = CDAP__INIT;
- invoke_id_t iid = key_to_invoke_id(key);
- struct cdap_rcvd * rcvd = cdap_rcvd_get_by_iid(instance, iid);
- if (rcvd == NULL)
- return -EINVAL;
+ struct cdap_rcvd * rcvd;
if (instance == NULL)
return -EINVAL;
+ rcvd = cdap_rcvd_get_by_key(instance, key);
+ if (rcvd == NULL)
+ return -1;
+
msg.opcode = CDAP_REPLY;
- msg.invoke_id = iid;
+ msg.invoke_id = rcvd->iid;
msg.has_result = true;
msg.result = result;
@@ -774,7 +812,11 @@ int cdap_reply_send(struct cdap * instance,
fd = rcvd->fd;
+ release_id(instance, rcvd->key);
+
+ assert(rcvd->data == NULL);
assert(rcvd->name == NULL);
+ assert(rcvd->proc);
free(rcvd);
diff --git a/src/lib/cdap_req.c b/src/lib/cdap_req.c
index b60e73ad..0b77f266 100644
--- a/src/lib/cdap_req.c
+++ b/src/lib/cdap_req.c
@@ -30,8 +30,9 @@
#include <stdlib.h>
#include <assert.h>
-struct cdap_req * cdap_req_create(int fd,
- cdap_key_t key)
+struct cdap_req * cdap_req_create(int fd,
+ invoke_id_t iid,
+ cdap_key_t key)
{
struct cdap_req * creq = malloc(sizeof(*creq));
pthread_condattr_t cattr;
@@ -40,6 +41,7 @@ struct cdap_req * cdap_req_create(int fd,
return NULL;
creq->fd = fd;
+ creq->iid = iid;
creq->key = key;
creq->state = REQ_INIT;
creq->response = -1;
diff --git a/src/lib/cdap_req.h b/src/lib/cdap_req.h
index fe8e3613..592d26a0 100644
--- a/src/lib/cdap_req.h
+++ b/src/lib/cdap_req.h
@@ -31,6 +31,8 @@
#include <pthread.h>
+typedef cdap_key_t invoke_id_t;
+
enum creq_state {
REQ_NULL = 0,
REQ_INIT,
@@ -46,6 +48,7 @@ struct cdap_req {
int fd;
struct timespec birth;
cdap_key_t key;
+ invoke_id_t iid;
int response;
buffer_t data;
@@ -55,8 +58,9 @@ struct cdap_req {
pthread_mutex_t lock;
};
-struct cdap_req * cdap_req_create(int fd,
- cdap_key_t key);
+struct cdap_req * cdap_req_create(int fd,
+ cdap_key_t key,
+ invoke_id_t iid);
void cdap_req_destroy(struct cdap_req * creq);
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-16 23:11:09 +0000