corosync  2.4.5
ipc_glue.c
Go to the documentation of this file.
1 /*
2  * Copyright (c) 2010-2012 Red Hat, Inc.
3  *
4  * All rights reserved.
5  *
6  * Author: Angus Salkeld <asalkeld@redhat.com>
7  *
8  * This software licensed under BSD license, the text of which follows:
9  *
10  * Redistribution and use in source and binary forms, with or without
11  * modification, are permitted provided that the following conditions are met:
12  *
13  * - Redistributions of source code must retain the above copyright notice,
14  * this list of conditions and the following disclaimer.
15  * - Redistributions in binary form must reproduce the above copyright notice,
16  * this list of conditions and the following disclaimer in the documentation
17  * and/or other materials provided with the distribution.
18  * - Neither the name of Red Hat, Inc. nor the names of its
19  * contributors may be used to endorse or promote products derived from this
20  * software without specific prior written permission.
21  *
22  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
23  * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
24  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
25  * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
26  * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
27  * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
28  * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
29  * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
30  * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
31  * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF
32  * THE POSSIBILITY OF SUCH DAMAGE.
33  */
34 
35 #include <config.h>
36 
37 #include <stdlib.h>
38 #include <stdio.h>
39 #include <errno.h>
40 #include <assert.h>
41 #include <sys/uio.h>
42 #include <string.h>
43 
44 #include <qb/qbdefs.h>
45 #include <qb/qblist.h>
46 #include <qb/qbutil.h>
47 #include <qb/qbloop.h>
48 #include <qb/qbipcs.h>
49 
50 #include <corosync/swab.h>
51 #include <corosync/corotypes.h>
52 #include <corosync/corodefs.h>
53 #include <corosync/totem/totempg.h>
54 #include <corosync/logsys.h>
55 #include <corosync/icmap.h>
56 
57 #include "sync.h"
58 #include "timer.h"
59 #include "main.h"
60 #include "util.h"
61 #include "apidef.h"
62 #include "service.h"
63 
64 LOGSYS_DECLARE_SUBSYS ("MAIN");
65 
66 static struct corosync_api_v1 *api = NULL;
67 static int32_t ipc_not_enough_fds_left = 0;
68 static int32_t ipc_fc_is_quorate; /* boolean */
69 static int32_t ipc_fc_totem_queue_level; /* percentage used */
70 static int32_t ipc_fc_sync_in_process; /* boolean */
71 static int32_t ipc_allow_connections = 0; /* boolean */
72 
73 #define CS_IPCS_MAPPER_SERV_NAME 256
74 
76  int32_t id;
77  qb_ipcs_service_t *inst;
79 };
80 
81 struct outq_item {
82  void *msg;
83  size_t mlen;
84  struct list_head list;
85 };
86 
87 static struct cs_ipcs_mapper ipcs_mapper[SERVICES_COUNT_MAX];
88 
89 static int32_t cs_ipcs_job_add(enum qb_loop_priority p, void *data, qb_loop_job_dispatch_fn fn);
90 static int32_t cs_ipcs_dispatch_add(enum qb_loop_priority p, int32_t fd, int32_t events,
91  void *data, qb_ipcs_dispatch_fn_t fn);
92 static int32_t cs_ipcs_dispatch_mod(enum qb_loop_priority p, int32_t fd, int32_t events,
93  void *data, qb_ipcs_dispatch_fn_t fn);
94 static int32_t cs_ipcs_dispatch_del(int32_t fd);
95 static void outq_flush (void *data);
96 
97 
98 static struct qb_ipcs_poll_handlers corosync_poll_funcs = {
99  .job_add = cs_ipcs_job_add,
100  .dispatch_add = cs_ipcs_dispatch_add,
101  .dispatch_mod = cs_ipcs_dispatch_mod,
102  .dispatch_del = cs_ipcs_dispatch_del,
103 };
104 
105 static int32_t cs_ipcs_connection_accept (qb_ipcs_connection_t *c, uid_t euid, gid_t egid);
106 static void cs_ipcs_connection_created(qb_ipcs_connection_t *c);
107 static int32_t cs_ipcs_msg_process(qb_ipcs_connection_t *c,
108  void *data, size_t size);
109 static int32_t cs_ipcs_connection_closed (qb_ipcs_connection_t *c);
110 static void cs_ipcs_connection_destroyed (qb_ipcs_connection_t *c);
111 
112 static struct qb_ipcs_service_handlers corosync_service_funcs = {
113  .connection_accept = cs_ipcs_connection_accept,
114  .connection_created = cs_ipcs_connection_created,
115  .msg_process = cs_ipcs_msg_process,
116  .connection_closed = cs_ipcs_connection_closed,
117  .connection_destroyed = cs_ipcs_connection_destroyed,
118 };
119 
120 static const char* cs_ipcs_serv_short_name(int32_t service_id)
121 {
122  const char *name;
123  switch (service_id) {
124  case CFG_SERVICE:
125  name = "cfg";
126  break;
127  case CPG_SERVICE:
128  name = "cpg";
129  break;
130  case QUORUM_SERVICE:
131  name = "quorum";
132  break;
133  case PLOAD_SERVICE:
134  name = "pload";
135  break;
136  case VOTEQUORUM_SERVICE:
137  name = "votequorum";
138  break;
139  case MON_SERVICE:
140  name = "mon";
141  break;
142  case WD_SERVICE:
143  name = "wd";
144  break;
145  case CMAP_SERVICE:
146  name = "cmap";
147  break;
148  default:
149  name = NULL;
150  break;
151  }
152  return name;
153 }
154 
155 void cs_ipc_allow_connections(int32_t allow)
156 {
157  ipc_allow_connections = allow;
158 }
159 
160 int32_t cs_ipcs_service_destroy(int32_t service_id)
161 {
162  if (ipcs_mapper[service_id].inst) {
163  qb_ipcs_destroy(ipcs_mapper[service_id].inst);
164  ipcs_mapper[service_id].inst = NULL;
165  }
166  return 0;
167 }
168 
169 static int32_t cs_ipcs_connection_accept (qb_ipcs_connection_t *c, uid_t euid, gid_t egid)
170 {
171  int32_t service = qb_ipcs_service_id_get(c);
172  uint8_t u8;
173  char key_name[ICMAP_KEYNAME_MAXLEN];
174 
175  if (!ipc_allow_connections) {
176  log_printf(LOGSYS_LEVEL_DEBUG, "Denied connection, corosync is not ready");
177  return -EAGAIN;
178  }
179 
180  if (corosync_service[service] == NULL ||
181  ipcs_mapper[service].inst == NULL) {
182  return -ENOSYS;
183  }
184 
185  if (ipc_not_enough_fds_left) {
186  return -EMFILE;
187  }
188 
189  if (euid == 0 || egid == 0) {
190  return 0;
191  }
192 
193  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "uidgid.uid.%u", euid);
194  if (icmap_get_uint8(key_name, &u8) == CS_OK && u8 == 1)
195  return 0;
196 
197  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "uidgid.config.uid.%u", euid);
198  if (icmap_get_uint8(key_name, &u8) == CS_OK && u8 == 1)
199  return 0;
200 
201  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "uidgid.gid.%u", egid);
202  if (icmap_get_uint8(key_name, &u8) == CS_OK && u8 == 1)
203  return 0;
204 
205  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "uidgid.config.gid.%u", egid);
206  if (icmap_get_uint8(key_name, &u8) == CS_OK && u8 == 1)
207  return 0;
208 
209  log_printf(LOGSYS_LEVEL_ERROR, "Denied connection attempt from %d:%d", euid, egid);
210 
211  return -EACCES;
212 }
213 
214 static char * pid_to_name (pid_t pid, char *out_name, size_t name_len)
215 {
216  char *name;
217  char *rest;
218  FILE *fp;
219  char fname[32];
220  char buf[256];
221 
222  snprintf (fname, 32, "/proc/%d/stat", pid);
223  fp = fopen (fname, "r");
224  if (!fp) {
225  return NULL;
226  }
227 
228  if (fgets (buf, sizeof (buf), fp) == NULL) {
229  fclose (fp);
230  return NULL;
231  }
232  fclose (fp);
233 
234  name = strrchr (buf, '(');
235  if (!name) {
236  return NULL;
237  }
238 
239  /* move past the bracket */
240  name++;
241 
242  rest = strrchr (buf, ')');
243 
244  if (rest == NULL || rest[1] != ' ') {
245  return NULL;
246  }
247 
248  *rest = '\0';
249  /* move past the NULL and space */
250  rest += 2;
251 
252  /* copy the name */
253  strncpy (out_name, name, name_len - 1);
254  out_name[name_len - 1] = '\0';
255  return out_name;
256 }
257 
259  char *icmap_path;
261  int32_t queuing;
262  uint32_t queued;
263  uint64_t invalid_request;
264  uint64_t overload;
265  uint32_t sent;
266  char data[1];
267 };
268 
269 static void cs_ipcs_connection_created(qb_ipcs_connection_t *c)
270 {
271  int32_t service = 0;
272  struct cs_ipcs_conn_context *context;
273  char proc_name[32];
274  struct qb_ipcs_connection_stats stats;
275  int32_t size = sizeof(struct cs_ipcs_conn_context);
276  char key_name[ICMAP_KEYNAME_MAXLEN];
277  int set_client_pid = 0;
278  int set_proc_name = 0;
279 
280  log_printf(LOG_DEBUG, "connection created");
281 
282  service = qb_ipcs_service_id_get(c);
283 
284  size += corosync_service[service]->private_data_size;
285  context = calloc(1, size);
286  if (context == NULL) {
287  qb_ipcs_disconnect(c);
288  return;
289  }
290 
291  list_init(&context->outq_head);
292  context->queuing = QB_FALSE;
293  context->queued = 0;
294  context->sent = 0;
295 
296  qb_ipcs_context_set(c, context);
297 
298  if (corosync_service[service]->lib_init_fn(c) != 0) {
299  log_printf(LOG_ERR, "lib_init_fn failed, disconnecting");
300  qb_ipcs_disconnect(c);
301  return;
302  }
303  icmap_inc("runtime.connections.active");
304 
305  qb_ipcs_connection_stats_get(c, &stats, QB_FALSE);
306 
307  if (stats.client_pid > 0) {
308  if (pid_to_name (stats.client_pid, proc_name, sizeof(proc_name))) {
309  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "runtime.connections.%s:%u:%p",
310  proc_name, stats.client_pid, c);
311  set_proc_name = 1;
312  } else {
313  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "runtime.connections.%u:%p",
314  stats.client_pid, c);
315  }
316  set_client_pid = 1;
317  } else {
318  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "runtime.connections.%p", c);
319  }
320 
322 
323  context->icmap_path = strdup(key_name);
324  if (context->icmap_path == NULL) {
325  qb_ipcs_disconnect(c);
326  return;
327  }
328 
329  if (set_proc_name) {
330  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.name", context->icmap_path);
331  icmap_set_string(key_name, proc_name);
332  }
333 
334  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.client_pid", context->icmap_path);
335  if (set_client_pid) {
336  icmap_set_uint32(key_name, stats.client_pid);
337  } else {
338  icmap_set_uint32(key_name, 0);
339  }
340 
341  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.service_id", context->icmap_path);
342  icmap_set_uint32(key_name, service);
343 
344  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.responses", context->icmap_path);
345  icmap_set_uint64(key_name, 0);
346 
347  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.dispatched", context->icmap_path);
348  icmap_set_uint64(key_name, 0);
349 
350  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.requests", context->icmap_path);
351  icmap_set_uint64(key_name, 0);
352 
353  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.send_retries", context->icmap_path);
354  icmap_set_uint64(key_name, 0);
355 
356  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.recv_retries", context->icmap_path);
357  icmap_set_uint64(key_name, 0);
358 
359  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.flow_control", context->icmap_path);
360  icmap_set_uint32(key_name, 0);
361 
362  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.flow_control_count", context->icmap_path);
363  icmap_set_uint64(key_name, 0);
364 
365  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.queue_size", context->icmap_path);
366  icmap_set_uint32(key_name, 0);
367 
368  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.invalid_request", context->icmap_path);
369  icmap_set_uint64(key_name, 0);
370 
371  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.overload", context->icmap_path);
372  icmap_set_uint64(key_name, 0);
373 }
374 
375 void cs_ipc_refcnt_inc(void *conn)
376 {
377  qb_ipcs_connection_ref(conn);
378 }
379 
380 void cs_ipc_refcnt_dec(void *conn)
381 {
382  qb_ipcs_connection_unref(conn);
383 }
384 
385 void *cs_ipcs_private_data_get(void *conn)
386 {
387  struct cs_ipcs_conn_context *cnx;
388  cnx = qb_ipcs_context_get(conn);
389  return &cnx->data[0];
390 }
391 
392 static void cs_ipcs_connection_destroyed (qb_ipcs_connection_t *c)
393 {
394  struct cs_ipcs_conn_context *context;
395  struct list_head *list, *list_next;
396  struct outq_item *outq_item;
397 
398  log_printf(LOG_DEBUG, "%s() ", __func__);
399 
400  context = qb_ipcs_context_get(c);
401  if (context) {
402  for (list = context->outq_head.next;
403  list != &context->outq_head; list = list_next) {
404 
405  list_next = list->next;
406  outq_item = list_entry (list, struct outq_item, list);
407 
408  list_del (list);
409  free (outq_item->msg);
410  free (outq_item);
411  }
412  free(context);
413  }
414 }
415 
416 static int32_t cs_ipcs_connection_closed (qb_ipcs_connection_t *c)
417 {
418  int32_t res = 0;
419  int32_t service = qb_ipcs_service_id_get(c);
420  icmap_iter_t iter;
421  char prefix[ICMAP_KEYNAME_MAXLEN];
422  const char *key_name;
423  struct cs_ipcs_conn_context *cnx;
424 
425  log_printf(LOG_DEBUG, "%s() ", __func__);
426  res = corosync_service[service]->lib_exit_fn(c);
427  if (res != 0) {
428  return res;
429  }
430 
431  qb_loop_job_del(cs_poll_handle_get(), QB_LOOP_HIGH, c, outq_flush);
432 
433  cnx = qb_ipcs_context_get(c);
434 
435  snprintf(prefix, ICMAP_KEYNAME_MAXLEN, "%s.", cnx->icmap_path);
436  iter = icmap_iter_init(prefix);
437  while ((key_name = icmap_iter_next(iter, NULL, NULL)) != NULL) {
438  icmap_delete(key_name);
439  }
440  icmap_iter_finalize(iter);
441  free(cnx->icmap_path);
442 
443  icmap_inc("runtime.connections.closed");
444  icmap_dec("runtime.connections.active");
445 
446  return 0;
447 }
448 
450  const struct iovec *iov,
451  unsigned int iov_len)
452 {
453  int32_t rc = qb_ipcs_response_sendv(conn, iov, iov_len);
454  if (rc >= 0) {
455  return 0;
456  }
457  return rc;
458 }
459 
460 int cs_ipcs_response_send(void *conn, const void *msg, size_t mlen)
461 {
462  int32_t rc = qb_ipcs_response_send(conn, msg, mlen);
463  if (rc >= 0) {
464  return 0;
465  }
466  return rc;
467 }
468 
469 static void outq_flush (void *data)
470 {
471  qb_ipcs_connection_t *conn = data;
472  struct list_head *list, *list_next;
473  struct outq_item *outq_item;
474  int32_t rc;
475  struct cs_ipcs_conn_context *context = qb_ipcs_context_get(conn);
476 
477  for (list = context->outq_head.next;
478  list != &context->outq_head; list = list_next) {
479 
480  list_next = list->next;
481  outq_item = list_entry (list, struct outq_item, list);
482 
483  rc = qb_ipcs_event_send(conn, outq_item->msg, outq_item->mlen);
484  if (rc < 0 && rc != -EAGAIN) {
485  errno = -rc;
486  qb_perror(LOG_ERR, "qb_ipcs_event_send");
487  return;
488  } else if (rc == -EAGAIN) {
489  break;
490  }
491  assert(rc == outq_item->mlen);
492  context->sent++;
493  context->queued--;
494 
495  list_del (list);
496  free (outq_item->msg);
497  free (outq_item);
498  }
499  if (list_empty (&context->outq_head)) {
500  context->queuing = QB_FALSE;
501  log_printf(LOGSYS_LEVEL_INFO, "Q empty, queued:%d sent:%d.",
502  context->queued, context->sent);
503  context->queued = 0;
504  context->sent = 0;
505  } else {
506  qb_loop_job_add(cs_poll_handle_get(), QB_LOOP_HIGH, conn, outq_flush);
507  }
508 }
509 
510 static void msg_send_or_queue(qb_ipcs_connection_t *conn, const struct iovec *iov, uint32_t iov_len)
511 {
512  int32_t rc = 0;
513  int32_t i;
514  int32_t bytes_msg = 0;
515  struct outq_item *outq_item;
516  char *write_buf = 0;
517  struct cs_ipcs_conn_context *context = qb_ipcs_context_get(conn);
518 
519  for (i = 0; i < iov_len; i++) {
520  bytes_msg += iov[i].iov_len;
521  }
522 
523  if (!context->queuing) {
524  assert(list_empty (&context->outq_head));
525  rc = qb_ipcs_event_sendv(conn, iov, iov_len);
526  if (rc == bytes_msg) {
527  context->sent++;
528  return;
529  }
530  if (rc == -EAGAIN) {
531  context->queued = 0;
532  context->sent = 0;
533  context->queuing = QB_TRUE;
534  qb_loop_job_add(cs_poll_handle_get(), QB_LOOP_HIGH, conn, outq_flush);
535  } else {
536  log_printf(LOGSYS_LEVEL_ERROR, "event_send retuned %d, expected %d!", rc, bytes_msg);
537  return;
538  }
539  }
540  outq_item = malloc (sizeof (struct outq_item));
541  if (outq_item == NULL) {
542  qb_ipcs_disconnect(conn);
543  return;
544  }
545  outq_item->msg = malloc (bytes_msg);
546  if (outq_item->msg == NULL) {
547  free (outq_item);
548  qb_ipcs_disconnect(conn);
549  return;
550  }
551 
552  write_buf = outq_item->msg;
553  for (i = 0; i < iov_len; i++) {
554  memcpy (write_buf, iov[i].iov_base, iov[i].iov_len);
555  write_buf += iov[i].iov_len;
556  }
557  outq_item->mlen = bytes_msg;
558  list_init (&outq_item->list);
559  list_add_tail (&outq_item->list, &context->outq_head);
560  context->queued++;
561 }
562 
563 int cs_ipcs_dispatch_send(void *conn, const void *msg, size_t mlen)
564 {
565  struct iovec iov;
566  iov.iov_base = (void *)msg;
567  iov.iov_len = mlen;
568  msg_send_or_queue (conn, &iov, 1);
569  return 0;
570 }
571 
573  const struct iovec *iov,
574  unsigned int iov_len)
575 {
576  msg_send_or_queue(conn, iov, iov_len);
577  return 0;
578 }
579 
580 static int32_t cs_ipcs_msg_process(qb_ipcs_connection_t *c,
581  void *data, size_t size)
582 {
583  struct qb_ipc_response_header response;
584  struct qb_ipc_request_header *request_pt = (struct qb_ipc_request_header *)data;
585  int32_t service = qb_ipcs_service_id_get(c);
586  int32_t send_ok = 0;
587  int32_t is_async_call = QB_FALSE;
588  ssize_t res = -1;
589  int sending_allowed_private_data;
590  struct cs_ipcs_conn_context *cnx;
591 
592  send_ok = corosync_sending_allowed (service,
593  request_pt->id,
594  request_pt,
595  &sending_allowed_private_data);
596 
597  is_async_call = (service == CPG_SERVICE && request_pt->id == 2);
598 
599  /*
600  * This happens when the message contains some kind of invalid
601  * parameter, such as an invalid size
602  */
603  if (send_ok == -EINVAL) {
604  response.size = sizeof (response);
605  response.id = 0;
606  response.error = CS_ERR_INVALID_PARAM;
607 
608  cnx = qb_ipcs_context_get(c);
609  if (cnx) {
610  cnx->invalid_request++;
611  }
612 
613  if (is_async_call) {
614  log_printf(LOGSYS_LEVEL_INFO, "*** %s() invalid message! size:%d error:%d",
615  __func__, response.size, response.error);
616  } else {
617  qb_ipcs_response_send (c,
618  &response,
619  sizeof (response));
620  }
621  res = -EINVAL;
622  } else if (send_ok < 0) {
623  cnx = qb_ipcs_context_get(c);
624  if (cnx) {
625  cnx->overload++;
626  }
627  if (!is_async_call) {
628  /*
629  * Overload, tell library to retry
630  */
631  response.size = sizeof (response);
632  response.id = 0;
633  response.error = CS_ERR_TRY_AGAIN;
634  qb_ipcs_response_send (c,
635  &response,
636  sizeof (response));
637  } else {
639  "*** %s() (%d:%d - %d) %s!",
640  __func__, service, request_pt->id,
641  is_async_call, strerror(-send_ok));
642  }
643  res = -ENOBUFS;
644  }
645 
646  if (send_ok >= 0) {
647  corosync_service[service]->lib_engine[request_pt->id].lib_handler_fn(c, request_pt);
648  res = 0;
649  }
650  corosync_sending_allowed_release (&sending_allowed_private_data);
651  return res;
652 }
653 
654 
655 static int32_t cs_ipcs_job_add(enum qb_loop_priority p, void *data, qb_loop_job_dispatch_fn fn)
656 {
657  return qb_loop_job_add(cs_poll_handle_get(), p, data, fn);
658 }
659 
660 static int32_t cs_ipcs_dispatch_add(enum qb_loop_priority p, int32_t fd, int32_t events,
661  void *data, qb_ipcs_dispatch_fn_t fn)
662 {
663  return qb_loop_poll_add(cs_poll_handle_get(), p, fd, events, data, fn);
664 }
665 
666 static int32_t cs_ipcs_dispatch_mod(enum qb_loop_priority p, int32_t fd, int32_t events,
667  void *data, qb_ipcs_dispatch_fn_t fn)
668 {
669  return qb_loop_poll_mod(cs_poll_handle_get(), p, fd, events, data, fn);
670 }
671 
672 static int32_t cs_ipcs_dispatch_del(int32_t fd)
673 {
674  return qb_loop_poll_del(cs_poll_handle_get(), fd);
675 }
676 
677 static void cs_ipcs_low_fds_event(int32_t not_enough, int32_t fds_available)
678 {
679  ipc_not_enough_fds_left = not_enough;
680  if (not_enough) {
681  log_printf(LOGSYS_LEVEL_WARNING, "refusing new connections (fds_available:%d)",
682  fds_available);
683  } else {
684  log_printf(LOGSYS_LEVEL_NOTICE, "allowing new connections (fds_available:%d)",
685  fds_available);
686 
687  }
688 }
689 
690 int32_t cs_ipcs_q_level_get(void)
691 {
692  return ipc_fc_totem_queue_level;
693 }
694 
695 static qb_loop_timer_handle ipcs_check_for_flow_control_timer;
696 static void cs_ipcs_check_for_flow_control(void)
697 {
698  int32_t i;
699  int32_t fc_enabled;
700 
701  for (i = 0; i < SERVICES_COUNT_MAX; i++) {
702  if (corosync_service[i] == NULL || ipcs_mapper[i].inst == NULL) {
703  continue;
704  }
705  fc_enabled = QB_IPCS_RATE_OFF;
706  if (ipc_fc_is_quorate == 1 ||
707  corosync_service[i]->allow_inquorate == CS_LIB_ALLOW_INQUORATE) {
708  /*
709  * we are quorate
710  * now check flow control
711  */
712  if (ipc_fc_totem_queue_level != TOTEM_Q_LEVEL_CRITICAL &&
713  ipc_fc_sync_in_process == 0) {
714  fc_enabled = QB_FALSE;
715  } else if (ipc_fc_totem_queue_level != TOTEM_Q_LEVEL_CRITICAL &&
716  i == VOTEQUORUM_SERVICE) {
717  /*
718  * Allow message processing for votequorum service even
719  * in sync phase
720  */
721  fc_enabled = QB_FALSE;
722  } else {
723  fc_enabled = QB_IPCS_RATE_OFF_2;
724  }
725  }
726  if (fc_enabled) {
727  qb_ipcs_request_rate_limit(ipcs_mapper[i].inst, fc_enabled);
728 
729  qb_loop_timer_add(cs_poll_handle_get(), QB_LOOP_MED, 1*QB_TIME_NS_IN_MSEC,
730  NULL, corosync_recheck_the_q_level, &ipcs_check_for_flow_control_timer);
731  } else if (ipc_fc_totem_queue_level == TOTEM_Q_LEVEL_LOW) {
732  qb_ipcs_request_rate_limit(ipcs_mapper[i].inst, QB_IPCS_RATE_FAST);
733  } else if (ipc_fc_totem_queue_level == TOTEM_Q_LEVEL_GOOD) {
734  qb_ipcs_request_rate_limit(ipcs_mapper[i].inst, QB_IPCS_RATE_NORMAL);
735  } else if (ipc_fc_totem_queue_level == TOTEM_Q_LEVEL_HIGH) {
736  qb_ipcs_request_rate_limit(ipcs_mapper[i].inst, QB_IPCS_RATE_SLOW);
737  }
738  }
739 }
740 
741 static void cs_ipcs_fc_quorum_changed(int quorate, void *context)
742 {
743  ipc_fc_is_quorate = quorate;
744  cs_ipcs_check_for_flow_control();
745 }
746 
747 static void cs_ipcs_totem_queue_level_changed(enum totem_q_level level)
748 {
749  ipc_fc_totem_queue_level = level;
750  cs_ipcs_check_for_flow_control();
751 }
752 
753 void cs_ipcs_sync_state_changed(int32_t sync_in_process)
754 {
755  ipc_fc_sync_in_process = sync_in_process;
756  cs_ipcs_check_for_flow_control();
757 }
758 
760 {
761  int32_t i;
762  struct qb_ipcs_stats srv_stats;
763  struct qb_ipcs_connection_stats stats;
764  qb_ipcs_connection_t *c, *prev;
765  struct cs_ipcs_conn_context *cnx;
766  char key_name[ICMAP_KEYNAME_MAXLEN];
767 
768  for (i = 0; i < SERVICES_COUNT_MAX; i++) {
769  if (corosync_service[i] == NULL || ipcs_mapper[i].inst == NULL) {
770  continue;
771  }
772  qb_ipcs_stats_get(ipcs_mapper[i].inst, &srv_stats, QB_FALSE);
773 
774  for (c = qb_ipcs_connection_first_get(ipcs_mapper[i].inst);
775  c;
776  prev = c, c = qb_ipcs_connection_next_get(ipcs_mapper[i].inst, prev), qb_ipcs_connection_unref(prev)) {
777 
778  cnx = qb_ipcs_context_get(c);
779  if (cnx == NULL) continue;
780 
781  qb_ipcs_connection_stats_get(c, &stats, QB_FALSE);
782 
783  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.client_pid", cnx->icmap_path);
784  icmap_set_uint32(key_name, stats.client_pid);
785 
786  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.requests", cnx->icmap_path);
787  icmap_set_uint64(key_name, stats.requests);
788 
789  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.responses", cnx->icmap_path);
790  icmap_set_uint64(key_name, stats.responses);
791 
792  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.dispatched", cnx->icmap_path);
793  icmap_set_uint64(key_name, stats.events);
794 
795  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.send_retries", cnx->icmap_path);
796  icmap_set_uint64(key_name, stats.send_retries);
797 
798  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.recv_retries", cnx->icmap_path);
799  icmap_set_uint64(key_name, stats.recv_retries);
800 
801  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.flow_control", cnx->icmap_path);
802  icmap_set_uint32(key_name, stats.flow_control_state);
803 
804  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.flow_control_count", cnx->icmap_path);
805  icmap_set_uint64(key_name, stats.flow_control_count);
806 
807  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.queue_size", cnx->icmap_path);
808  icmap_set_uint32(key_name, cnx->queued);
809 
810  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.invalid_request", cnx->icmap_path);
811  icmap_set_uint64(key_name, cnx->invalid_request);
812 
813  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.overload", cnx->icmap_path);
814  icmap_set_uint64(key_name, cnx->overload);
815  }
816  }
817 }
818 
819 static enum qb_ipc_type cs_get_ipc_type (void)
820 {
821  char *str;
822  int found = 0;
823  enum qb_ipc_type ret = QB_IPC_NATIVE;
824 
825  if (icmap_get_string("qb.ipc_type", &str) != CS_OK) {
826  log_printf(LOGSYS_LEVEL_DEBUG, "No configured qb.ipc_type. Using native ipc");
827  return QB_IPC_NATIVE;
828  }
829 
830  if (strcmp(str, "native") == 0) {
831  ret = QB_IPC_NATIVE;
832  found = 1;
833  }
834 
835  if (strcmp(str, "shm") == 0) {
836  ret = QB_IPC_SHM;
837  found = 1;
838  }
839 
840  if (strcmp(str, "socket") == 0) {
841  ret = QB_IPC_SOCKET;
842  found = 1;
843  }
844 
845  if (found) {
846  log_printf(LOGSYS_LEVEL_DEBUG, "Using %s ipc", str);
847  } else {
848  log_printf(LOGSYS_LEVEL_DEBUG, "Unknown ipc type %s", str);
849  }
850 
851  free(str);
852 
853  return ret;
854 }
855 
856 const char *cs_ipcs_service_init(struct corosync_service_engine *service)
857 {
858  const char *serv_short_name;
859 
860  serv_short_name = cs_ipcs_serv_short_name(service->id);
861 
862  if (service->lib_engine_count == 0) {
864  "NOT Initializing IPC on %s [%d]",
865  serv_short_name,
866  service->id);
867  return NULL;
868  }
869 
870  if (strlen(serv_short_name) >= CS_IPCS_MAPPER_SERV_NAME) {
871  log_printf (LOGSYS_LEVEL_ERROR, "service name %s is too long", serv_short_name);
872  return "qb_ipcs_run error";
873  }
874 
875  ipcs_mapper[service->id].id = service->id;
876  strcpy(ipcs_mapper[service->id].name, serv_short_name);
878  "Initializing IPC on %s [%d]",
879  ipcs_mapper[service->id].name,
880  ipcs_mapper[service->id].id);
881  ipcs_mapper[service->id].inst = qb_ipcs_create(ipcs_mapper[service->id].name,
882  ipcs_mapper[service->id].id,
883  cs_get_ipc_type(),
884  &corosync_service_funcs);
885  assert(ipcs_mapper[service->id].inst);
886  qb_ipcs_poll_handlers_set(ipcs_mapper[service->id].inst,
887  &corosync_poll_funcs);
888  if (qb_ipcs_run(ipcs_mapper[service->id].inst) != 0) {
889  log_printf (LOGSYS_LEVEL_ERROR, "Can't initialize IPC");
890  return "qb_ipcs_run error";
891  }
892 
893  return NULL;
894 }
895 
896 void cs_ipcs_init(void)
897 {
898  api = apidef_get ();
899 
900  qb_loop_poll_low_fds_event_set(cs_poll_handle_get(), cs_ipcs_low_fds_event);
901 
902  api->quorum_register_callback (cs_ipcs_fc_quorum_changed, NULL);
903  totempg_queue_level_register_callback (cs_ipcs_totem_queue_level_changed);
904 
905  icmap_set_uint64("runtime.connections.active", 0);
906  icmap_set_uint64("runtime.connections.closed", 0);
907 }
908