Obsolete
/
libipc-old
Archived
3
0
Fork 0
This repository has been archived on 2024-06-18. You can view files and clone it, but cannot push or open issues/pull-requests.
libipc-old/lib/pubsubd.c

603 lines
14 KiB
C
Raw Normal View History

2016-06-05 20:48:13 +02:00
#include "pubsubd.h"
// WORKERS: one thread per client
void pubsubd_workers_init (struct workers *wrkrs) { LIST_INIT(wrkrs); }
struct worker *
pubsubd_workers_add (struct workers *wrkrs, const struct worker *w)
{
if (wrkrs == NULL || w == NULL) {
printf ("pubsubd_workers_add: wrkrs == NULL or w == NULL");
return NULL;
}
struct worker *n = malloc (sizeof (struct worker));
memset (n, 0, sizeof (struct worker));
memcpy (n, w, sizeof (struct worker));
if (w->ale != NULL)
n->ale = pubsubd_app_list_elm_copy (w->ale);
LIST_INSERT_HEAD(wrkrs, n, entries);
return n;
}
void pubsubd_worker_del (struct workers *wrkrs, struct worker *w)
{
struct worker *todel = pubsubd_worker_get (wrkrs, w);
if(todel != NULL) {
LIST_REMOVE(todel, entries);
pubsubd_worker_free (todel);
free (todel);
todel = NULL;
}
}
// kill the threads
void pubsubd_workers_stop (struct workers *wrkrs)
{
if (!wrkrs)
return;
struct worker *w = NULL;
struct worker *wtmp = NULL;
LIST_FOREACH_SAFE(w, wrkrs, entries, wtmp) {
if (w->thr == NULL)
continue;
pthread_cancel (*w->thr);
void *ret = NULL;
pthread_join (*w->thr, &ret);
if (ret != NULL) {
free (ret);
}
free (w->thr);
w->thr = NULL;
}
}
void pubsubd_workers_del_all (struct workers *wrkrs)
{
if (!wrkrs)
return;
struct worker *w = NULL;
while (!LIST_EMPTY(wrkrs)) {
printf ("KILL THE WORKERS : %p\n", w);
w = LIST_FIRST(wrkrs);
LIST_REMOVE(w, entries);
pubsubd_worker_free (w);
free (w);
w = NULL;
}
}
void pubsubd_worker_free (struct worker * w)
{
if (w == NULL)
return;
pubsubd_app_list_elm_free (w->ale);
free (w->ale);
w->ale = NULL;
}
struct worker * pubsubd_worker_get (struct workers *wrkrs, struct worker *w)
{
struct worker * np = NULL;
LIST_FOREACH(np, wrkrs, entries) {
if (pubsubd_worker_eq (np, w))
return np;
}
return NULL;
}
int pubsubd_worker_eq (const struct worker *w1, const struct worker *w2)
{
return w1 == w2; // if it's the same pointer
}
// a thread for each connected process
void * pubsubd_worker_thread (void *params)
{
int s = 0;
s = pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
if (s != 0)
printf ("pthread_setcancelstate: %d\n", s);
struct worker *w = (struct worker *) params;
if (w == NULL) {
fprintf (stderr, "error pubsubd_worker_thread : params NULL\n");
return NULL;
}
struct channels *chans = w->chans;
struct channel *chan = w->chan;
struct app_list_elm *ale = w->ale;
// main loop
while (1) {
struct pubsub_msg m;
memset (&m, 0, sizeof (struct pubsub_msg));
pubsubd_msg_recv (ale->p, &m);
if (m.type == PUBSUB_TYPE_DISCONNECT) {
printf ("process %d disconnecting...\n", ale->p->pid);
if ( 0 != pubsubd_subscriber_del (chan->alh, ale)) {
fprintf (stderr, "err : subscriber not registered\n");
}
break;
}
else {
struct channel *ch = pubsubd_channel_search (chans, chan->chan);
if (ch == NULL) {
printf ("CHAN NOT FOUND\n");
}
else {
printf ("what should be sent: ");
pubsubd_msg_print (&m);
printf ("send the message to:\t");
pubsubd_channel_print (ch);
pubsubd_msg_send (ch->alh, &m);
}
}
pubsubd_msg_free (&m);
}
pubsubd_app_list_elm_free (ale);
free (w->ale);
w->ale = NULL;
free (w->thr);
w->thr = NULL;
pubsubd_worker_del (w->my_workers, w);
pthread_exit (NULL);
}
2016-06-12 12:38:43 +02:00
2016-06-05 20:48:13 +02:00
// CHANNELS
void pubsubd_channels_init (struct channels *chans) { LIST_INIT(chans); }
2016-06-07 17:44:18 +02:00
struct channel *
2016-09-11 01:47:25 +02:00
pubsubd_channels_add (struct channels *chans, const char *chan)
2016-06-05 20:48:13 +02:00
{
2016-09-11 01:47:25 +02:00
if(chans == NULL || chan == NULL) {
printf ("pubsubd_channels_add: chans == NULL or chan == NULL");
2016-06-07 17:44:18 +02:00
return NULL;
2016-09-11 01:47:25 +02:00
}
struct channel *n = malloc (sizeof (struct channel));
2016-09-11 01:47:25 +02:00
memset (n, 0, sizeof (struct channel));
pubsubd_channel_new (n, chan);
2016-06-05 20:48:13 +02:00
LIST_INSERT_HEAD(chans, n, entries);
2016-06-07 17:44:18 +02:00
return n;
2016-06-05 20:48:13 +02:00
}
void
pubsubd_channels_del (struct channels *chans, struct channel *c)
{
struct channel *todel = pubsubd_channel_get (chans, c);
if(todel != NULL) {
pubsubd_channel_free (todel);
2016-06-07 17:44:18 +02:00
LIST_REMOVE(todel, entries);
2016-06-05 20:48:13 +02:00
free (todel);
todel = NULL;
}
}
2016-06-07 13:49:23 +02:00
void pubsubd_channels_del_all (struct channels *chans)
{
if (!chans)
return;
struct channel *c = NULL;
2016-06-07 13:49:23 +02:00
while (!LIST_EMPTY(chans)) {
c = LIST_FIRST(chans);
LIST_REMOVE(c, entries);
pubsubd_channel_free (c);
free (c);
c = NULL;
}
}
2016-06-05 20:48:13 +02:00
struct channel * pubsubd_channel_copy (struct channel *c)
{
2016-06-07 17:44:18 +02:00
if (c == NULL)
return NULL;
struct channel *copy = NULL;
2016-06-05 20:48:13 +02:00
copy = malloc (sizeof(struct channel));
memset (copy, 0, sizeof (struct channel));
2016-06-07 17:44:18 +02:00
2016-06-05 20:48:13 +02:00
memcpy (copy, c, sizeof(struct channel));
2016-06-07 17:44:18 +02:00
if (c->chan != NULL) {
copy->chan = malloc (c->chanlen +1);
memset (copy->chan, 0, c->chanlen +1);
2016-09-07 23:46:00 +02:00
memcpy (copy->chan, c->chan, c->chanlen);
2016-06-07 17:44:18 +02:00
copy->chanlen = c->chanlen;
}
2016-09-11 01:47:25 +02:00
else {
printf ("pubsubd_channel_copy: c->chan == NULL\n");
}
2016-06-07 17:44:18 +02:00
2016-06-05 20:48:13 +02:00
return copy;
}
2016-09-07 23:46:00 +02:00
int pubsubd_channel_new (struct channel *c, const char * name)
{
if (c == NULL) {
return 1;
}
2016-09-09 12:06:25 +02:00
size_t nlen = (strlen (name) > BUFSIZ) ? BUFSIZ : strlen (name);
2016-09-07 23:46:00 +02:00
if (c->chan == NULL)
2016-09-09 12:06:25 +02:00
c->chan = malloc (nlen +1);
2016-09-07 23:46:00 +02:00
2016-09-09 12:06:25 +02:00
memset (c->chan, 0, nlen +1);
2016-09-07 23:46:00 +02:00
memcpy (c->chan, name, nlen);
c->chanlen = nlen;
2016-09-07 23:46:00 +02:00
return 0;
}
void pubsubd_channel_free (struct channel * c)
{
2016-06-07 17:44:18 +02:00
if (c == NULL)
return;
if (c->chan != NULL) {
free (c->chan);
c->chan = NULL;
}
if (c->alh != NULL) {
pubsubd_subscriber_del_all (c->alh);
free (c->alh);
}
}
2016-09-11 01:47:25 +02:00
struct channel * pubsubd_channel_search (struct channels *chans, char *chan)
{
struct channel * np = NULL;
LIST_FOREACH(np, chans, entries) {
2016-09-11 02:18:12 +02:00
// TODO debug
2016-09-11 14:37:41 +02:00
// printf ("pubsubd_channel_search: %s (%ld) vs %s (%ld)\n"
// , np->chan, np->chanlen, chan, strlen(chan));
2016-09-11 02:18:12 +02:00
if (np->chanlen == strlen (chan)
&& strncmp (np->chan, chan, np->chanlen) == 0) {
2016-09-11 14:37:41 +02:00
// printf ("pubsubd_channel_search: FOUND\n");
2016-09-11 01:47:25 +02:00
return np;
2016-09-11 02:18:12 +02:00
}
2016-09-11 01:47:25 +02:00
}
return NULL;
}
2016-06-05 20:48:13 +02:00
struct channel * pubsubd_channel_get (struct channels *chans, struct channel *c)
{
struct channel * np = NULL;
LIST_FOREACH(np, chans, entries) {
if (pubsubd_channel_eq (np, c))
2016-06-05 20:48:13 +02:00
return np;
}
return NULL;
}
int
pubsubd_channel_eq (const struct channel *c1, const struct channel *c2)
2016-06-05 20:48:13 +02:00
{
2016-09-10 02:54:53 +02:00
return c1->chanlen == c2->chanlen &&
strncmp (c1->chan, c2->chan, c1->chanlen) == 0;
2016-06-05 20:48:13 +02:00
}
// SUBSCRIBER
2016-06-07 17:44:18 +02:00
void pubsubd_subscriber_init (struct app_list_head **chans) {
if (chans == NULL)
return;
2016-09-10 17:16:39 +02:00
if (*chans == NULL) {
2016-06-07 17:44:18 +02:00
*chans = malloc (sizeof(struct channels));
2016-09-10 17:16:39 +02:00
memset (*chans, 0, sizeof(struct channels));
}
2016-06-07 17:44:18 +02:00
LIST_INIT(*chans);
}
2016-06-05 20:48:13 +02:00
2016-09-11 14:37:41 +02:00
void pubsubd_channel_print (const struct channel *chan)
{
if (chan->chan == NULL) {
printf ("pubsubd_channels_print: chan->chan == NULL\n");
}
printf ( "\033[32mchan %s\033[00m\n", chan->chan);
if (chan->alh == NULL)
printf ("pubsubd_channels_print: chan->alh == NULL\n");
else
pubsubd_subscriber_print (chan->alh);
}
2016-06-07 17:44:18 +02:00
void pubsubd_channels_print (const struct channels *chans)
2016-06-07 13:49:23 +02:00
{
printf ("\033[36mmchannels\033[00m\n");
2016-06-07 17:44:18 +02:00
2016-09-11 01:47:25 +02:00
if (chans == NULL) {
// TODO debug
printf ("pubsubd_channels_print: chans == NULL\n");
2016-06-07 17:44:18 +02:00
return ;
2016-09-11 01:47:25 +02:00
}
2016-06-07 13:49:23 +02:00
2016-06-07 17:44:18 +02:00
struct channel *chan = NULL;
LIST_FOREACH(chan, chans, entries) {
2016-09-11 14:37:41 +02:00
pubsubd_channel_print (chan);
2016-06-07 17:44:18 +02:00
}
2016-06-07 13:49:23 +02:00
}
struct app_list_elm * pubsubd_app_list_elm_copy (const struct app_list_elm *ale)
2016-06-05 20:48:13 +02:00
{
if (ale == NULL)
return NULL;
struct app_list_elm * n = NULL;
2016-06-05 20:48:13 +02:00
n = malloc (sizeof (struct app_list_elm));
2016-09-10 17:16:39 +02:00
memset (n, 0, sizeof (struct app_list_elm));
2016-06-05 20:48:13 +02:00
2016-06-07 17:44:18 +02:00
if (ale->p != NULL)
n->p = srv_process_copy (ale->p);
2016-06-05 20:48:13 +02:00
2016-06-08 17:57:05 +02:00
n->action = ale->action;
2016-06-05 20:48:13 +02:00
return n;
}
int
pubsubd_subscriber_eq (const struct app_list_elm *ale1, const struct app_list_elm *ale2)
{
return srv_process_eq (ale1->p, ale2->p);
}
2016-06-05 20:48:13 +02:00
void
pubsubd_subscriber_add (struct app_list_head *alh, const struct app_list_elm *ale)
2016-06-05 20:48:13 +02:00
{
2016-09-11 01:47:25 +02:00
if(alh == NULL || ale == NULL) {
fprintf (stderr, "err alh or ale is NULL\n");
2016-06-05 20:48:13 +02:00
return;
2016-09-11 01:47:25 +02:00
}
2016-06-05 20:48:13 +02:00
struct app_list_elm *n = pubsubd_app_list_elm_copy (ale);
LIST_INSERT_HEAD(alh, n, entries);
}
struct app_list_elm *
2016-06-13 09:47:19 +02:00
pubsubd_subscriber_get (const struct app_list_head *alh, const struct app_list_elm *p)
2016-06-05 20:48:13 +02:00
{
struct app_list_elm *np = NULL, *res = NULL;
2016-06-13 09:47:19 +02:00
LIST_FOREACH(np, alh, entries) {
if(pubsubd_subscriber_eq (np, p)) {
2016-06-05 20:48:13 +02:00
res = np;
}
}
return res;
}
2016-09-11 01:47:25 +02:00
void pubsubd_subscriber_print (struct app_list_head *alh)
{
struct app_list_elm *np = NULL;
LIST_FOREACH(np, alh, entries) {
printf ("\t");
srv_process_print (np->p);
}
}
2016-06-13 09:47:19 +02:00
int
pubsubd_subscriber_del (struct app_list_head *alh, struct app_list_elm *p)
2016-06-05 20:48:13 +02:00
{
2016-06-13 09:47:19 +02:00
struct app_list_elm *todel = pubsubd_subscriber_get (alh, p);
2016-06-05 20:48:13 +02:00
if(todel != NULL) {
pubsubd_app_list_elm_free (todel);
2016-06-07 17:44:18 +02:00
LIST_REMOVE(todel, entries);
2016-06-05 20:48:13 +02:00
free (todel);
todel = NULL;
2016-06-13 09:47:19 +02:00
return 0;
2016-06-05 20:48:13 +02:00
}
2016-06-13 09:47:19 +02:00
return 1;
2016-06-05 20:48:13 +02:00
}
2016-06-07 17:44:18 +02:00
void pubsubd_subscriber_del_all (struct app_list_head *alh)
{
if (!alh)
return;
struct app_list_elm *ale = NULL;
2016-06-07 17:44:18 +02:00
while (!LIST_EMPTY(alh)) {
ale = LIST_FIRST(alh);
LIST_REMOVE(ale, entries);
pubsubd_app_list_elm_free (ale);
free (ale);
ale = NULL;
}
}
2016-06-05 20:48:13 +02:00
void pubsubd_app_list_elm_create (struct app_list_elm *ale, struct process *p)
{
if (ale == NULL)
return;
2016-09-10 17:16:39 +02:00
if (ale->p != NULL)
free (ale->p);
2016-06-05 20:48:13 +02:00
ale->p = srv_process_copy (p);
}
void pubsubd_app_list_elm_free (struct app_list_elm *todel)
{
2016-06-07 17:44:18 +02:00
if (todel == NULL || todel->p == NULL)
return;
2016-09-09 12:06:25 +02:00
free (todel->p);
2016-06-05 20:48:13 +02:00
}
2016-09-09 21:52:56 +02:00
int pubsubd_get_new_process (const char *spath, struct app_list_elm *ale
2016-06-08 17:57:05 +02:00
, struct channels *chans, struct channel **c)
2016-06-05 20:48:13 +02:00
{
if (spath == NULL || ale == NULL || chans == NULL) {
fprintf (stderr, "pubsubd_get_new_process: spath or ale or chans == NULL\n");
2016-06-07 17:44:18 +02:00
return -1;
}
2016-06-07 17:44:18 +02:00
char *buf = NULL;
size_t msize = 0;
2016-09-09 21:52:56 +02:00
file_read (spath, &buf, &msize);
2016-06-07 11:45:21 +02:00
// parse pubsubd init msg (sent in TMPDIR/<service>)
//
// line fmt : pid index version action chan
2016-06-07 17:44:18 +02:00
// action : quit | pub | sub
2016-06-07 11:45:21 +02:00
size_t i = 0;
char *str = NULL, *token = NULL, *saveptr = NULL;
2016-06-07 11:45:21 +02:00
pid_t pid = 0;
int index = 0;
int version = 0;
2016-06-07 11:45:21 +02:00
// chan name
2016-06-07 17:44:18 +02:00
char chan[BUFSIZ];
memset (chan, 0, BUFSIZ);
2016-06-07 17:44:18 +02:00
2016-09-10 17:16:39 +02:00
if (buf == NULL) {
return -2;
}
2016-09-09 01:36:44 +02:00
printf ("INIT: %s\n", buf);
2016-06-07 11:45:21 +02:00
for (str = buf, i = 1; ; str = NULL, i++) {
token = strtok_r(str, " ", &saveptr);
if (token == NULL)
break;
switch (i) {
case 1 : pid = strtoul(token, NULL, 10); break;
case 2 : index = strtoul(token, NULL, 10); break;
case 3 : version = strtoul(token, NULL, 10); break;
2016-06-07 17:44:18 +02:00
case 4 : {
if (strncmp("both", token, 4) == 0) {
ale->action = PUBSUB_BOTH;
2016-06-07 11:45:21 +02:00
}
2016-06-07 17:44:18 +02:00
else if (strncmp("pub", token, 3) == 0) {
ale->action = PUBSUB_PUB;
}
2016-06-07 11:45:21 +02:00
else if (strncmp("sub", token, 3) == 0) {
2016-06-07 17:44:18 +02:00
ale->action = PUBSUB_SUB;
2016-06-07 13:49:23 +02:00
}
else { // everything else is about killing the service
2016-06-07 17:44:18 +02:00
ale->action = PUBSUB_QUIT;
2016-06-07 11:45:21 +02:00
}
break;
}
case 5 : {
2016-09-09 21:52:56 +02:00
// for the last element of the line
// drop the following \n
if (ale->action != PUBSUB_QUIT) {
memcpy (chan, token, (strlen (token) < BUFSIZ) ?
2016-09-09 01:36:44 +02:00
strlen (token) -1 : BUFSIZ);
}
break;
2016-06-07 11:45:21 +02:00
}
}
}
2016-06-07 17:44:18 +02:00
if (buf != NULL) {
free (buf);
buf = NULL;
}
2016-06-08 17:57:05 +02:00
if (ale->action == PUBSUB_QUIT) {
return 0;
2016-06-07 17:44:18 +02:00
}
if (ale->p != NULL) {
free (ale->p);
ale->p = NULL;
}
2016-06-08 17:57:05 +02:00
ale->p = malloc (sizeof (struct process));
2016-09-10 17:16:39 +02:00
memset (ale->p, 0, sizeof (struct process));
2016-06-08 17:57:05 +02:00
srv_process_gen (ale->p, pid, index, version);
chan[BUFSIZ -1] = '\0';
2016-09-07 23:46:00 +02:00
2016-09-11 01:47:25 +02:00
// not found = new
struct channel *new_chan = NULL;
2016-09-11 01:47:25 +02:00
new_chan = pubsubd_channel_search (chans, chan);
2016-06-08 17:57:05 +02:00
if (new_chan == NULL) {
2016-09-11 01:47:25 +02:00
new_chan = pubsubd_channels_add (chans, chan);
2016-06-08 17:57:05 +02:00
pubsubd_subscriber_init (&new_chan->alh);
}
*c = new_chan;
2016-06-07 17:44:18 +02:00
// add the subscriber
2016-09-11 01:47:25 +02:00
if (ale->action == PUBSUB_SUB || ale->action == PUBSUB_BOTH) {
printf ("new process in chan %s\n", chan);
pubsubd_subscriber_add ((*c)->alh, ale);
}
2016-06-07 11:45:21 +02:00
return 0;
2016-06-05 20:48:13 +02:00
}
2016-06-07 11:45:21 +02:00
2016-06-08 17:57:05 +02:00
// alh from the channel, message to send
2016-06-07 11:45:21 +02:00
void pubsubd_msg_send (const struct app_list_head *alh, const struct pubsub_msg * m)
{
2016-09-11 14:37:41 +02:00
if (alh == NULL) {
fprintf (stderr, "pubsubd_msg_send: alh == NULL");
return;
}
if (m == NULL) {
fprintf (stderr, "pubsubd_msg_send: m == NULL");
return;
}
2016-06-08 17:57:05 +02:00
struct app_list_elm * ale = NULL;
char *buf = NULL;
size_t msize = 0;
2016-06-08 17:57:05 +02:00
pubsubd_msg_serialize (m, &buf, &msize);
LIST_FOREACH(ale, alh, entries) {
srv_write (ale->p, buf, msize);
}
if (buf != NULL) {
free (buf);
}
}
2016-06-07 11:45:21 +02:00
void pubsubd_msg_recv (struct process *p, struct pubsub_msg *m)
{
// read the message from the process
size_t mlen = 0;
char *buf = NULL;
2016-09-11 14:37:41 +02:00
while (buf == NULL || mlen == 0) {
srv_read (p, &buf, &mlen);
}
2016-06-07 11:45:21 +02:00
pubsubd_msg_unserialize (m, buf, mlen);
2016-06-08 17:57:05 +02:00
if (buf != NULL) {
free (buf);
}
2016-06-05 20:48:13 +02:00
}