Proxmox-Port/libqb
1
0
Fork 0
mirror of https://salsa.debian.org/ha-team/libqb synced 2026-08-11 19:12:02 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

IPC: get the poll independent functions working.

Browse Source 
Signed-off-by: Angus Salkeld <asalkeld@redhat.com>
This commit is contained in:
Angus Salkeld 2010-10-09 21:13:32 +11:00
parent 1f6c42b098
commit 8df21458f5
9 changed files with 107 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
include/qb/qbipcs.h
View File

@ -26,6 +26,7 @@
#include <stdlib.h>
#include <qb/qbipc_common.h>
#include <qb/qbhdb.h>
#include <qb/qbloop.h>
/* *INDENT-OFF* */
#ifdef __cplusplus
@ -38,17 +39,26 @@ typedef struct qb_ipcs_connection qb_ipcs_connection_t;
typedef qb_handle_t qb_ipcs_service_pt;
typedef int32_t (*qb_ipcs_dispatch_fn_t) (qb_ipcs_service_pt s, int32_t fd, int32_t revents,
typedef int32_t (*qb_ipcs_dispatch_fn_t) (int32_t fd, int32_t revents,
void *data);
typedef int32_t (*qb_ipcs_dispatch_add_fn)(qb_ipcs_service_pt s, int32_t fd, int32_t events,
void *data, qb_ipcs_dispatch_fn_t fn);
typedef int32_t (*qb_ipcs_dispatch_rm_fn)(qb_ipcs_service_pt s, int32_t fd, int32_t events,
void *data, qb_ipcs_dispatch_fn_t fn);
typedef int32_t (*qb_ipcs_dispatch_add_fn)(enum qb_loop_priority p,
int32_t fd,
int32_t events,
void *data,
qb_ipcs_dispatch_fn_t fn);
typedef int32_t (*qb_ipcs_dispatch_mod_fn)(enum qb_loop_priority p,
int32_t fd,
int32_t events,
void *data,
qb_ipcs_dispatch_fn_t fn);
typedef int32_t (*qb_ipcs_dispatch_del_fn)(int32_t fd);
struct qb_ipcs_poll_handlers {
qb_ipcs_dispatch_add_fn dispatch_add;
qb_ipcs_dispatch_rm_fn dispatch_rm;
qb_ipcs_dispatch_mod_fn dispatch_mod;
qb_ipcs_dispatch_del_fn dispatch_del;
};
/**
@ -102,7 +112,7 @@ void qb_ipcs_poll_handlers_set(qb_ipcs_service_pt s,
/**
* run the new IPC server.
*/
int32_t qb_ipcs_run(qb_ipcs_service_pt s, void *loop_pt);
int32_t qb_ipcs_run(qb_ipcs_service_pt s);
/**
* Destroy the IPC server.
@ -145,6 +155,14 @@ void qb_ipcs_context_set(qb_ipcs_connection_t *c, void *context);
void *qb_ipcs_context_get(qb_ipcs_connection_t *c);
enum qb_ipcs_rate_limit {
QB_IPCS_RATE_FAST,
QB_IPCS_RATE_NORMAL,
QB_IPCS_RATE_SLOW,
QB_IPCS_RATE_OFF,
};
void qb_ipcs_request_rate_limit(enum qb_ipcs_rate_limit rl);
/* *INDENT-OFF* */
#ifdef __cplusplus

9
include/qb/qbloop.h
View File

@ -88,10 +88,11 @@ int32_t qb_loop_poll_add(qb_loop_t *l,
qb_loop_poll_dispatch_fn dispatch_fn);
int32_t qb_loop_poll_mod(qb_loop_t *l,
enum qb_loop_priority p,
int32_t fd,
int32_t events,
qb_loop_poll_dispatch_fn dispatch_fn);
enum qb_loop_priority p,
int32_t fd,
int32_t events,
void *data,
qb_loop_poll_dispatch_fn dispatch_fn);
int32_t qb_loop_poll_del(qb_loop_t *l, int32_t fd);

3
lib/hdb.c
View File

@ -283,8 +283,7 @@ int32_t qb_hdb_handle_refcount_get(struct qb_hdb * hdb, qb_handle_t handle_in)
return (refcount);
}
void qb_hdb_iterator_reset(struct qb_hdb
*hdb)
void qb_hdb_iterator_reset(struct qb_hdb *hdb)
{
hdb->iterator = 0;
}

3
lib/ipc_int.h
View File

@ -28,6 +28,7 @@
#include <dirent.h>
#include <mqueue.h>
#include <qb/qblist.h>
#include <qb/qbloop.h>
#include <qb/qbipcc.h>
#include <qb/qbipcs.h>
#include <qb/qbipc_common.h>
@ -147,11 +148,11 @@ struct qb_ipcs_service {
pid_t pid;
int32_t needs_sock_for_poll;
int32_t server_sock;
void* loop_pt;
struct qb_ipcs_service_handlers serv_fns;
struct qb_ipcs_poll_handlers poll_fns;
struct qb_ipcs_funcs funcs;
enum qb_loop_priority poll_priority;
struct qb_list_head connections;
};

3
lib/ipc_posix_mq.c
View File

@ -379,8 +379,7 @@ static int32_t qb_ipcs_pmq_connect(struct qb_ipcs_service *s,
}
if (!s->needs_sock_for_poll) {
qb_loop_poll_add(s->loop_pt, QB_LOOP_HIGH,
c->request.u.pmq.q,
s->poll_fns.dispatch_add(s->poll_priority, c->request.u.pmq.q,
POLLIN | POLLPRI | POLLNVAL,
c, qb_ipcs_dispatch_service_request);
}

10
lib/ipc_us.c
View File

@ -520,7 +520,7 @@ int32_t qb_ipcs_us_publish(struct qb_ipcs_service * s)
qb_util_log(LOG_ERR, "listen failed: %s.\n", error_str);
}
qb_loop_poll_add(s->loop_pt, QB_LOOP_MED, s->server_sock,
s->poll_fns.dispatch_add(s->poll_priority, s->server_sock,
POLLIN | POLLPRI | POLLNVAL,
s, qb_ipcs_us_connection_acceptor);
return 0;
@ -598,10 +598,10 @@ retry_accept:
c->receive_buf = malloc(c->request.max_msg_size);
if (s->needs_sock_for_poll) {
qb_loop_poll_add(s->loop_pt, QB_LOOP_MED, c->sock,
POLLIN | POLLPRI | POLLNVAL,
c,
qb_ipcs_dispatch_connection_request);
s->poll_fns.dispatch_add(s->poll_priority, c->sock,
POLLIN | POLLPRI | POLLNVAL,
c,
qb_ipcs_dispatch_connection_request);
}
}

46
lib/ipcs.c
View File

@ -44,6 +44,7 @@ qb_ipcs_service_pt qb_ipcs_create(const char *name,
s->pid = getpid();
s->type = type;
s->needs_sock_for_poll = QB_FALSE;
s->poll_priority = QB_LOOP_MED;
s->service_id = service_id;
strncpy(s->name, name, NAME_MAX);
@ -68,20 +69,19 @@ void qb_ipcs_poll_handlers_set(qb_ipcs_service_pt pt,
qb_hdb_handle_get(&qb_ipc_services, pt, (void **)&s);
s->poll_fns.dispatch_add = handlers->dispatch_add;
s->poll_fns.dispatch_rm = handlers->dispatch_rm;
s->poll_fns.dispatch_mod = handlers->dispatch_mod;
s->poll_fns.dispatch_del = handlers->dispatch_del;
qb_hdb_handle_put(&qb_ipc_services, pt);
}
int32_t qb_ipcs_run(qb_ipcs_service_pt pt, void *loop_pt)
int32_t qb_ipcs_run(qb_ipcs_service_pt pt)
{
int32_t res;
struct qb_ipcs_service *s;
qb_hdb_handle_get(&qb_ipc_services, pt, (void **)&s);
s->loop_pt = loop_pt;
res = qb_ipcs_us_publish(s);
if (res < 0) {
qb_hdb_handle_put(&qb_ipc_services, pt);
@ -332,3 +332,41 @@ void *qb_ipcs_context_get(struct qb_ipcs_connection *c)
return c->context;
}
void qb_ipcs_request_rate_limit(enum qb_ipcs_rate_limit rl)
{
struct qb_ipcs_service *s;
struct qb_ipcs_connection *c;
qb_handle_t handle;
enum qb_loop_priority p;
switch (rl) {
case QB_IPCS_RATE_SLOW: p = QB_LOOP_LOW; break;
case QB_IPCS_RATE_NORMAL: p = QB_LOOP_MED; break;
case QB_IPCS_RATE_FAST: p = QB_LOOP_HIGH; break;
default:
case QB_IPCS_RATE_OFF:
assert(0);
break;
}
qb_hdb_iterator_reset(&qb_ipc_services);
while (qb_hdb_iterator_next(&qb_ipc_services, (void**)&s, &handle)) {
qb_list_for_each_entry(c, &s->connections, list) {
if (s->type == QB_IPC_POSIX_MQ && !s->needs_sock_for_poll) {
s->poll_fns.dispatch_mod(p, c->request.u.pmq.q,
POLLIN | POLLPRI | POLLNVAL,
c, qb_ipcs_dispatch_service_request);
} else {
s->poll_fns.dispatch_mod(p, c->sock,
POLLIN | POLLPRI | POLLNVAL,
c,
qb_ipcs_dispatch_connection_request);
}
}
s->poll_priority = p;
qb_hdb_handle_put(&qb_ipc_services, handle);
}
}

1
lib/loop_poll.c
View File

@ -360,6 +360,7 @@ int32_t qb_loop_poll_mod(struct qb_loop *l,
enum qb_loop_priority p,
int32_t fd,
int32_t events,
void *data,
qb_loop_poll_dispatch_fn dispatch_fn)
{
int32_t i;

26
tests/bms.c
View File

@ -137,6 +137,23 @@ static void show_usage(const char *name)
printf("\n");
}
static int32_t my_dispatch_add(enum qb_loop_priority p, int32_t fd, int32_t events,
void *data, qb_ipcs_dispatch_fn_t fn)
{
return qb_loop_poll_add(bms_loop, p, fd, events, data, fn);
}
static int32_t my_dispatch_mod(enum qb_loop_priority p, int32_t fd, int32_t events,
void *data, qb_ipcs_dispatch_fn_t fn)
{
return qb_loop_poll_mod(bms_loop, p, fd, events, data, fn);
}
static int32_t my_dispatch_del(int32_t fd)
{
return qb_loop_poll_del(bms_loop, fd);
}
int32_t main(int32_t argc, char *argv[])
{
const char *options = "nvhmps";
@ -148,6 +165,11 @@ int32_t main(int32_t argc, char *argv[])
.msg_process = s1_msg_process_fn,
.connection_destroyed = s1_connection_destroyed_fn,
};
struct qb_ipcs_poll_handlers ph = {
.dispatch_add = my_dispatch_add,
.dispatch_mod = my_dispatch_mod,
.dispatch_del = my_dispatch_del,
};
while ((opt = getopt(argc, argv, options)) != -1) {
switch (opt) {
@ -186,7 +208,9 @@ int32_t main(int32_t argc, char *argv[])
perror("qb_ipcs_create");
exit(1);
}
qb_ipcs_run(s1, bms_loop);
qb_ipcs_poll_handlers_set(s1, &ph);
qb_ipcs_run(s1);
qb_loop_run(bms_loop);
return EXIT_SUCCESS;