Archived
3
0

unix socket rewrite: done

This commit is contained in:
Philippe PITTOLI 2018-10-08 16:15:35 +02:00
parent fefd301279
commit 340ff653d6
14 changed files with 75 additions and 124 deletions

View File

@ -20,15 +20,15 @@ int main()
} }
} }
ipc_client_array_print(&clients); ipc_clients_print(&clients);
ret = ipc_client_del(&clients, &client_tab[2]); ret = ipc_client_del(&clients, &client_tab[2]);
if(ret < 0) { if(ret < 0) {
printf("erreur %d\n", ret ); printf("erreur %d\n", ret );
} }
ipc_client_array_print(&clients); ipc_clients_print(&clients);
ipc_client_array_free (&clients); ipc_clients_free (&clients);
return 0; return 0;
} }

View File

@ -34,6 +34,7 @@ int ipc_client_add (struct ipc_clients *clients, struct ipc_client *p)
{ {
assert(clients != NULL); assert(clients != NULL);
assert(p != NULL); assert(p != NULL);
clients->size++; clients->size++;
clients->clients = realloc(clients->clients clients->clients = realloc(clients->clients
, sizeof(struct ipc_client) * clients->size); , sizeof(struct ipc_client) * clients->size);
@ -62,7 +63,7 @@ int ipc_client_del (struct ipc_clients *clients, struct ipc_client *p)
clients->clients[i] = clients->clients[clients->size-1]; clients->clients[i] = clients->clients[clients->size-1];
clients->size--; clients->size--;
if (clients->size == 0) { if (clients->size == 0) {
ipc_client_array_free (clients); ipc_clients_free (clients);
} }
else { else {
clients->clients = realloc(clients->clients clients->clients = realloc(clients->clients
@ -87,7 +88,7 @@ void client_print (struct ipc_client *p)
, p->proc_fd, p->index, p->version); , p->proc_fd, p->index, p->version);
} }
void ipc_client_array_print (struct ipc_clients *ap) void ipc_clients_print (struct ipc_clients *ap)
{ {
int i; int i;
for (i = 0; i < ap->size; i++) { for (i = 0; i < ap->size; i++) {
@ -96,7 +97,7 @@ void ipc_client_array_print (struct ipc_clients *ap)
} }
} }
void ipc_client_array_free (struct ipc_clients *ap) void ipc_clients_free (struct ipc_clients *ap)
{ {
if (ap->clients != NULL) { if (ap->clients != NULL) {
free (ap->clients); free (ap->clients);

View File

@ -12,11 +12,12 @@ struct ipc_clients {
int size; int size;
}; };
// store and remove only pointers on allocated structures
int ipc_client_add (struct ipc_clients *, struct ipc_client *); int ipc_client_add (struct ipc_clients *, struct ipc_client *);
int ipc_client_del (struct ipc_clients *clients, struct ipc_client *p); int ipc_client_del (struct ipc_clients *, struct ipc_client *);
void ipc_client_array_print (struct ipc_clients *); void ipc_clients_print (struct ipc_clients *);
void ipc_client_array_free (struct ipc_clients *); void ipc_clients_free (struct ipc_clients *);
struct ipc_client * ipc_server_client_copy (const struct ipc_client *p); struct ipc_client * ipc_server_client_copy (const struct ipc_client *p);
int ipc_server_client_eq (const struct ipc_client *p1, const struct ipc_client *p2); int ipc_server_client_eq (const struct ipc_client *p1, const struct ipc_client *p2);

View File

@ -53,39 +53,20 @@ int ipc_server_accept (struct ipc_service *srv, struct ipc_client *p)
return -1; return -1;
} }
#if 0
struct ipc_message m_con;
memset (&m_con, 0, sizeof (struct ipc_message));
ipc_server_read (p, &m_con);
// TODO: handle the parameters in the first message
ipc_message_free (&m_con);
struct ipc_message m_ack;
memset (&m_ack, 0, sizeof (struct ipc_message));
ipc_message_format_ack (&m_ack, NULL, 0);
ipc_server_write (p, &m_ack);
ipc_message_free (&m_ack);
#endif
return 0; return 0;
} }
// empty the srv structure
int ipc_server_close (struct ipc_service *srv) int ipc_server_close (struct ipc_service *srv)
{ {
usock_close (srv->service_fd); usock_close (srv->service_fd);
return usock_remove (srv->spath); int ret = usock_remove (srv->spath);
ipc_service_empty (srv);
return ret;
} }
int ipc_server_close_client (struct ipc_client *p) int ipc_server_close_client (struct ipc_client *p)
{ {
// struct ipc_message m_ack_dis;
// memset (&m_ack_dis, 0, sizeof (struct ipc_message));
// m_ack_dis.type = MSG_TYPE_CLOSE;
// if (ipc_message_write (p->proc_fd, &m_ack_dis) < 0) {
// handle_err ("server_close_client", "ipc_message_write < 0");
// }
return usock_close (p->proc_fd); return usock_close (p->proc_fd);
} }
@ -100,8 +81,7 @@ int ipc_server_write (const struct ipc_client *p, const struct ipc_message *m)
} }
int ipc_application_connection (int argc, char **argv, char **env int ipc_application_connection (int argc, char **argv, char **env
, struct ipc_service *srv, const char *sname , struct ipc_service *srv, const char *sname)
, const char *connectionstr, size_t msize)
{ {
argc = argc; argc = argc;
argv = argv; argv = argv;
@ -123,48 +103,14 @@ int ipc_application_connection (int argc, char **argv, char **env
return -1; return -1;
} }
#if 0
// send connection string and receive acknowledgement
struct ipc_message m_con;
memset (&m_con, 0, sizeof (struct ipc_message));
// format the connection msg
if (ipc_message_format_con (&m_con, connectionstr, msize) < 0) {
handle_err ("application_connection", "ipc_message_format_con");
return -1;
}
// send connection msg
ipc_application_write (srv, &m_con);
ipc_message_free (&m_con);
// receive ack msg
struct ipc_message m_ack;
memset (&m_ack, 0, sizeof (struct ipc_message));
ipc_application_read (srv, &m_ack);
assert (m_ack.type == MSG_TYPE_ACK);
assert (m_ack.length == 0);
ipc_message_free (&m_ack);
#endif
return 0; return 0;
} }
#if 0
// send a CLOSE message then close the socket // send a CLOSE message then close the socket
int ipc_application_close (struct ipc_service *srv) int ipc_application_close (struct ipc_service *srv)
{ {
struct ipc_message m;
memset (&m, 0, sizeof (struct ipc_message));
m.type = MSG_TYPE_CLOSE;
if (ipc_message_write (srv->service_fd, &m) < 0) {
handle_err ("application_close", "ipc_message_write < 0");
}
return usock_close (srv->service_fd); return usock_close (srv->service_fd);
} }
#endif
int ipc_application_read (struct ipc_service *srv, struct ipc_message *m) int ipc_application_read (struct ipc_service *srv, struct ipc_message *m)
{ {
@ -211,7 +157,7 @@ int ipc_server_select (struct ipc_clients *clients, struct ipc_service *srv
assert (active_clients != NULL); assert (active_clients != NULL);
// delete previous read active_clients array // delete previous read active_clients array
ipc_client_array_free (active_clients); ipc_clients_free (active_clients);
int i, j; int i, j;
/* master file descriptor list */ /* master file descriptor list */

View File

@ -37,7 +37,7 @@ int ipc_server_select (struct ipc_clients * clients, struct ipc_service *srv
// send the connection string to $TMP/<service> // send the connection string to $TMP/<service>
// fill srv->spath && srv->service_fd // fill srv->spath && srv->service_fd
int ipc_application_connection (int argc, char **argv, char **env int ipc_application_connection (int argc, char **argv, char **env
, struct ipc_service *, const char *, const char *, size_t); , struct ipc_service *, const char *);
int ipc_application_close (struct ipc_service *); int ipc_application_close (struct ipc_service *);
int ipc_application_read (struct ipc_service *srv, struct ipc_message *m); int ipc_application_read (struct ipc_service *srv, struct ipc_message *m);

View File

@ -10,7 +10,7 @@ void ipc_message_print (const struct ipc_message *m)
printf ("msg: type %d len %d\n", m->type, m->length); printf ("msg: type %d len %d\n", m->type, m->length);
} }
int ipc_message_format_read (struct ipc_message *m, const char *buf, size_t msize) int ipc_message_format_read (struct ipc_message *m, const char *buf, ssize_t msize)
{ {
assert (m != NULL); assert (m != NULL);
assert (buf != NULL); assert (buf != NULL);
@ -37,7 +37,7 @@ int ipc_message_format_read (struct ipc_message *m, const char *buf, size_t msiz
return 0; return 0;
} }
int ipc_message_format_write (const struct ipc_message *m, char **buf, size_t *msize) int ipc_message_format_write (const struct ipc_message *m, char **buf, ssize_t *msize)
{ {
assert (m != NULL); assert (m != NULL);
assert (buf != NULL); assert (buf != NULL);
@ -79,7 +79,7 @@ int ipc_message_read (int fd, struct ipc_message *m)
assert (m != NULL); assert (m != NULL);
char *buf = NULL; char *buf = NULL;
size_t msize = BUFSIZ; ssize_t msize = BUFSIZ;
int ret = usock_recv (fd, &buf, &msize); int ret = usock_recv (fd, &buf, &msize);
if (ret < 0) { if (ret < 0) {
@ -134,7 +134,7 @@ int ipc_message_write (int fd, const struct ipc_message *m)
// MSG FORMAT // MSG FORMAT
int ipc_message_format (struct ipc_message *m, char type, const char *payload, size_t length) int ipc_message_format (struct ipc_message *m, char type, const char *payload, ssize_t length)
{ {
assert (m != NULL); assert (m != NULL);
assert (length + 3 <= BUFSIZ); assert (length + 3 <= BUFSIZ);
@ -166,23 +166,11 @@ int ipc_message_format (struct ipc_message *m, char type, const char *payload, s
return 0; return 0;
} }
int ipc_message_format_data (struct ipc_message *m, const char *payload, size_t length) int ipc_message_format_data (struct ipc_message *m, const char *payload, ssize_t length)
{ {
return ipc_message_format (m, MSG_TYPE_DATA, payload, length); return ipc_message_format (m, MSG_TYPE_DATA, payload, length);
} }
#if 0
int ipc_message_format_con (struct ipc_message *m, const char *payload, size_t length)
{
return ipc_message_format (m, MSG_TYPE_CON, payload, length);
}
int ipc_message_format_ack (struct ipc_message *m, const char *payload, size_t length)
{
return ipc_message_format (m, MSG_TYPE_ACK, payload, length);
}
#endif
int ipc_message_format_server_close (struct ipc_message *m) int ipc_message_format_server_close (struct ipc_message *m)
{ {
return ipc_message_format (m, MSG_TYPE_SERVER_CLOSE, NULL, 0); return ipc_message_format (m, MSG_TYPE_SERVER_CLOSE, NULL, 0);

View File

@ -20,9 +20,9 @@ struct ipc_message {
}; };
// used to create msg structure from buffer // used to create msg structure from buffer
int ipc_message_format_read (struct ipc_message *m, const char *buf, size_t msize); int ipc_message_format_read (struct ipc_message *m, const char *buf, ssize_t msize);
// used to create buffer from msg structure // used to create buffer from msg structure
int ipc_message_format_write (const struct ipc_message *m, char **buf, size_t *msize); int ipc_message_format_write (const struct ipc_message *m, char **buf, ssize_t *msize);
// read a structure msg from fd // read a structure msg from fd
// 1 on a recipient socket close // 1 on a recipient socket close
@ -30,9 +30,7 @@ int ipc_message_read (int fd, struct ipc_message *m);
// write a structure msg to fd // write a structure msg to fd
int ipc_message_write (int fd, const struct ipc_message *m); int ipc_message_write (int fd, const struct ipc_message *m);
// int ipc_message_format_con (struct ipc_message *m, const char *payload, size_t length); int ipc_message_format_data (struct ipc_message *m, const char *payload, ssize_t length);
int ipc_message_format_data (struct ipc_message *m, const char *payload, size_t length);
// int ipc_message_format_ack (struct ipc_message *m, const char *payload, size_t length);
int ipc_message_format_server_close (struct ipc_message *m); int ipc_message_format_server_close (struct ipc_message *m);
int ipc_message_empty (struct ipc_message *m); int ipc_message_empty (struct ipc_message *m);

View File

@ -246,13 +246,6 @@ int usock_close (int fd)
{ {
int ret = 0; int ret = 0;
ret = fsync (fd);
if (ret < 0) {
handle_err ("usock_close", "fsync ret < 0");
perror ("closing");
return -1;
}
ret = close (fd); ret = close (fd);
if (ret < 0) { if (ret < 0) {
handle_err ("usock_close", "close ret < 0"); handle_err ("usock_close", "close ret < 0");

View File

@ -48,4 +48,6 @@ int usock_accept (int fd, int *pfd);
// same as unlink(2) // same as unlink(2)
int usock_remove (const char *path); int usock_remove (const char *path);
static inline int ipc_service_empty (struct ipc_service *srv) { srv = srv; return 0 ;};
#endif #endif

View File

@ -22,7 +22,7 @@ void non_interactive (int argc, char *argv[], char *env[])
srv.version = 0; srv.version = 0;
// init service // init service
if (ipc_application_connection (argc, argv, env, &srv, SERVICE_NAME, NULL, 0) < 0) { if (ipc_application_connection (argc, argv, env, &srv, SERVICE_NAME) < 0) {
handle_err("main", "server_init < 0"); handle_err("main", "server_init < 0");
exit (EXIT_FAILURE); exit (EXIT_FAILURE);
} }
@ -35,7 +35,7 @@ void non_interactive (int argc, char *argv[], char *env[])
handle_err("main", "application_write < 0"); handle_err("main", "application_write < 0");
exit (EXIT_FAILURE); exit (EXIT_FAILURE);
} }
ipc_message_free (&m); ipc_message_empty (&m);
if (ipc_application_read (&srv, &m) < 0) { if (ipc_application_read (&srv, &m) < 0) {
handle_err("main", "application_read < 0"); handle_err("main", "application_read < 0");
@ -43,7 +43,7 @@ void non_interactive (int argc, char *argv[], char *env[])
} }
printf ("msg recv: %s\n", m.payload); printf ("msg recv: %s\n", m.payload);
ipc_message_free (&m); ipc_message_empty (&m);
if (ipc_application_close (&srv) < 0) { if (ipc_application_close (&srv) < 0) {
handle_err("main", "application_close < 0"); handle_err("main", "application_close < 0");
@ -69,7 +69,7 @@ void interactive (int argc, char *argv[], char *env[])
srv.version = 0; srv.version = 0;
// init service // init service
if (ipc_application_connection (argc, argv, env, &srv, SERVICE_NAME, NULL, 0) < 0) { if (ipc_application_connection (argc, argv, env, &srv, SERVICE_NAME) < 0) {
handle_err ("main", "server_init < 0"); handle_err ("main", "server_init < 0");
exit (EXIT_FAILURE); exit (EXIT_FAILURE);
} }
@ -96,7 +96,7 @@ void interactive (int argc, char *argv[], char *env[])
handle_err("main", "application_write < 0"); handle_err("main", "application_write < 0");
exit (EXIT_FAILURE); exit (EXIT_FAILURE);
} }
ipc_message_free (&m); ipc_message_empty (&m);
if (ipc_application_read (&srv, &m) < 0) { if (ipc_application_read (&srv, &m) < 0) {
handle_err("main", "application_read < 0"); handle_err("main", "application_read < 0");
@ -104,7 +104,7 @@ void interactive (int argc, char *argv[], char *env[])
} }
printf ("msg recv: %s", m.payload); printf ("msg recv: %s", m.payload);
ipc_message_free (&m); ipc_message_empty (&m);
} }
if (ask_server_to_quit) { if (ask_server_to_quit) {
@ -114,7 +114,7 @@ void interactive (int argc, char *argv[], char *env[])
handle_err("main", "application_write < 0"); handle_err("main", "application_write < 0");
exit (EXIT_FAILURE); exit (EXIT_FAILURE);
} }
ipc_message_free (&m); ipc_message_empty (&m);
} else if (ipc_application_close (&srv) < 0) { } else if (ipc_application_close (&srv) < 0) {
handle_err("main", "application_close < 0"); handle_err("main", "application_close < 0");
exit (EXIT_FAILURE); exit (EXIT_FAILURE);

View File

@ -39,37 +39,63 @@ void handle_new_msg (struct ipc_clients *clients, struct ipc_clients *clients_ta
int i = 0; int i = 0;
for (i = 0; i < clients_talking->size; i++) { for (i = 0; i < clients_talking->size; i++) {
// printf ("loop handle_new_msg\n"); // printf ("loop handle_new_msg\n");
ret = ipc_server_read (clients_talking->clients[i], &m);
// current talking client
struct ipc_client *pc = clients_talking->clients[i];
ret = ipc_server_read (pc, &m);
if (ret < 0) { if (ret < 0) {
handle_error("server_read < 0"); handle_error("server_read < 0");
} }
// close the client then delete it from the client array // close the client then delete it from clients
if (ret == 1) { if (ret == 1) {
cpt--; cpt--;
printf ("disconnection => %d client(s) remaining\n", cpt); printf ("disconnection => %d client(s) remaining\n", cpt);
if (ipc_server_close_client (clients_talking->clients[i]) < 0) if (ipc_server_close_client (pc) < 0)
handle_err( "handle_new_msg", "server_close_client < 0"); handle_err( "handle_new_msg", "server_close_client < 0");
if (ipc_client_del (clients, clients_talking->clients[i]) < 0) if (ipc_client_del (clients, pc) < 0)
handle_err( "handle_new_msg", "ipc_client_del < 0"); handle_err( "handle_new_msg", "ipc_client_del < 0");
if (ipc_client_del (clients_talking, clients_talking->clients[i]) < 0) if (ipc_client_del (clients_talking, pc) < 0)
handle_err( "handle_new_msg", "ipc_client_del < 0"); handle_err( "handle_new_msg", "ipc_client_del < 0");
i--; i--;
// free the ipc_client structure
free (pc);
continue; continue;
} }
if (m.type == MSG_TYPE_SERVER_CLOSE) { if (m.type == MSG_TYPE_SERVER_CLOSE) {
// free remaining clients
for (int y = 0; y < clients->size ; y++) {
struct ipc_client *cli = clients->clients[y];
// TODO: replace with specific ipc_client_empty function
if (cli != NULL)
free (cli);
clients->clients[y] = NULL;
}
ipc_clients_free (clients);
ipc_clients_free (clients_talking);
if (ipc_server_close (srv) < 0) { if (ipc_server_close (srv) < 0) {
handle_error("server_close < 0"); handle_error("server_close < 0");
} }
ipc_message_empty (&m);
free (srv);
exit (0); exit (0);
} }
printf ("new message : %s", m.payload); printf ("new message : %s", m.payload);
if (ipc_server_write (clients_talking->clients[i], &m) < 0) { if (ipc_server_write (pc, &m) < 0) {
handle_err( "handle_new_msg", "server_write < 0"); handle_err( "handle_new_msg", "server_write < 0");
} }
// empty the message structure
ipc_message_empty (&m);
memset (&m, 0, sizeof m);
} }
} }
@ -83,7 +109,7 @@ void handle_new_msg (struct ipc_clients *clients, struct ipc_clients *clients_ta
void main_loop () void main_loop ()
{ {
int i, ret = 0; int ret = 0;
struct ipc_clients clients; struct ipc_clients clients;
memset(&clients, 0, sizeof(struct ipc_clients)); memset(&clients, 0, sizeof(struct ipc_clients));
@ -105,14 +131,10 @@ void main_loop ()
if (clients_talking.size > 0) { if (clients_talking.size > 0) {
handle_new_msg (&clients, &clients_talking); handle_new_msg (&clients, &clients_talking);
} }
ipc_client_array_free (&clients_talking); ipc_clients_free (&clients_talking);
}
for (i = 0; i < clients.size; i++) {
if (ipc_server_close_client (clients.clients[i]) < 0) {
handle_error( "server_close_client < 0");
}
} }
// should never go there
exit (1);
} }

View File

@ -19,7 +19,7 @@ void pubsubd_channel_print (const struct channel *chan)
handle_err ("pubsubd_channel_print", "chan->subs == NULL"); handle_err ("pubsubd_channel_print", "chan->subs == NULL");
} }
else { else {
ipc_client_array_print (chan->subs); ipc_clients_print (chan->subs);
} }
} }
@ -115,7 +115,7 @@ void pubsubd_channel_free (struct channel * c)
} }
if (c->subs != NULL) { if (c->subs != NULL) {
ipc_client_array_free (c->subs); ipc_clients_free (c->subs);
free (c->subs); free (c->subs);
} }
} }

View File

@ -186,7 +186,7 @@ void pubsubd_main_loop (struct ipc_service *srv, struct channels *chans)
handle_new_connection (srv, &ap); handle_new_connection (srv, &ap);
handle_new_msg (chans, &ap, &proc_to_read); handle_new_msg (chans, &ap, &proc_to_read);
} }
ipc_client_array_free (&proc_to_read); ipc_clients_free (&proc_to_read);
} }
for (i = 0; i < ap.size; i++) { for (i = 0; i < ap.size; i++) {

View File

@ -132,7 +132,7 @@ void remoted_main_loop (struct ipc_service *srv, struct remoted_ctx *ctx)
handle_new_connection (srv, &ap); handle_new_connection (srv, &ap);
handle_new_msg (&ap, &proc_to_read); handle_new_msg (&ap, &proc_to_read);
} }
ipc_client_array_free (&proc_to_read); ipc_clients_free (&proc_to_read);
} }
for (i = 0; i < ap.size; i++) { for (i = 0; i < ap.size; i++) {