corosync  2.3.5-dirty
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.gid.%u", egid);
198  if (icmap_get_uint8(key_name, &u8) == CS_OK && u8 == 1)
199  return 0;
200 
201  log_printf(LOGSYS_LEVEL_ERROR, "Denied connection attempt from %d:%d", euid, egid);
202 
203  return -EACCES;
204 }
205 
206 static char * pid_to_name (pid_t pid, char *out_name, size_t name_len)
207 {
208  char *name;
209  char *rest;
210  FILE *fp;
211  char fname[32];
212  char buf[256];
213 
214  snprintf (fname, 32, "/proc/%d/stat", pid);
215  fp = fopen (fname, "r");
216  if (!fp) {
217  return NULL;
218  }
219 
220  if (fgets (buf, sizeof (buf), fp) == NULL) {
221  fclose (fp);
222  return NULL;
223  }
224  fclose (fp);
225 
226  name = strrchr (buf, '(');
227  if (!name) {
228  return NULL;
229  }
230 
231  /* move past the bracket */
232  name++;
233 
234  rest = strrchr (buf, ')');
235 
236  if (rest == NULL || rest[1] != ' ') {
237  return NULL;
238  }
239 
240  *rest = '\0';
241  /* move past the NULL and space */
242  rest += 2;
243 
244  /* copy the name */
245  strncpy (out_name, name, name_len);
246  out_name[name_len - 1] = '\0';
247  return out_name;
248 }
249 
251  char *icmap_path;
253  int32_t queuing;
254  uint32_t queued;
255  uint64_t invalid_request;
256  uint64_t overload;
257  uint32_t sent;
258  char data[1];
259 };
260 
261 static void cs_ipcs_connection_created(qb_ipcs_connection_t *c)
262 {
263  int32_t service = 0;
264  struct cs_ipcs_conn_context *context;
265  char proc_name[32];
266  struct qb_ipcs_connection_stats stats;
267  int32_t size = sizeof(struct cs_ipcs_conn_context);
268  char key_name[ICMAP_KEYNAME_MAXLEN];
269  int set_client_pid = 0;
270  int set_proc_name = 0;
271 
272  log_printf(LOG_DEBUG, "connection created");
273 
274  service = qb_ipcs_service_id_get(c);
275 
276  size += corosync_service[service]->private_data_size;
277  context = calloc(1, size);
278  if (context == NULL) {
279  qb_ipcs_disconnect(c);
280  return;
281  }
282 
283  list_init(&context->outq_head);
284  context->queuing = QB_FALSE;
285  context->queued = 0;
286  context->sent = 0;
287 
288  qb_ipcs_context_set(c, context);
289 
290  if (corosync_service[service]->lib_init_fn(c) != 0) {
291  log_printf(LOG_ERR, "lib_init_fn failed, disconnecting");
292  qb_ipcs_disconnect(c);
293  return;
294  }
295  icmap_inc("runtime.connections.active");
296 
297  qb_ipcs_connection_stats_get(c, &stats, QB_FALSE);
298 
299  if (stats.client_pid > 0) {
300  if (pid_to_name (stats.client_pid, proc_name, sizeof(proc_name))) {
301  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "runtime.connections.%s:%u:%p",
302  proc_name, stats.client_pid, c);
303  set_proc_name = 1;
304  } else {
305  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "runtime.connections.%u:%p",
306  stats.client_pid, c);
307  }
308  set_client_pid = 1;
309  } else {
310  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "runtime.connections.%p", c);
311  }
312 
314 
315  context->icmap_path = strdup(key_name);
316  if (context->icmap_path == NULL) {
317  qb_ipcs_disconnect(c);
318  return;
319  }
320 
321  if (set_proc_name) {
322  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.name", context->icmap_path);
323  icmap_set_string(key_name, proc_name);
324  }
325 
326  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.client_pid", context->icmap_path);
327  if (set_client_pid) {
328  icmap_set_uint32(key_name, stats.client_pid);
329  } else {
330  icmap_set_uint32(key_name, 0);
331  }
332 
333  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.service_id", context->icmap_path);
334  icmap_set_uint32(key_name, service);
335 
336  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.responses", context->icmap_path);
337  icmap_set_uint64(key_name, 0);
338 
339  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.dispatched", context->icmap_path);
340  icmap_set_uint64(key_name, 0);
341 
342  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.requests", context->icmap_path);
343  icmap_set_uint64(key_name, 0);
344 
345  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.send_retries", context->icmap_path);
346  icmap_set_uint64(key_name, 0);
347 
348  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.recv_retries", context->icmap_path);
349  icmap_set_uint64(key_name, 0);
350 
351  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.flow_control", context->icmap_path);
352  icmap_set_uint32(key_name, 0);
353 
354  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.flow_control_count", context->icmap_path);
355  icmap_set_uint64(key_name, 0);
356 
357  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.queue_size", context->icmap_path);
358  icmap_set_uint32(key_name, 0);
359 
360  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.invalid_request", context->icmap_path);
361  icmap_set_uint64(key_name, 0);
362 
363  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.overload", context->icmap_path);
364  icmap_set_uint64(key_name, 0);
365 }
366 
367 void cs_ipc_refcnt_inc(void *conn)
368 {
369  qb_ipcs_connection_ref(conn);
370 }
371 
372 void cs_ipc_refcnt_dec(void *conn)
373 {
374  qb_ipcs_connection_unref(conn);
375 }
376 
377 void *cs_ipcs_private_data_get(void *conn)
378 {
379  struct cs_ipcs_conn_context *cnx;
380  cnx = qb_ipcs_context_get(conn);
381  return &cnx->data[0];
382 }
383 
384 static void cs_ipcs_connection_destroyed (qb_ipcs_connection_t *c)
385 {
386  struct cs_ipcs_conn_context *context;
387  struct list_head *list, *list_next;
388  struct outq_item *outq_item;
389 
390  log_printf(LOG_DEBUG, "%s() ", __func__);
391 
392  context = qb_ipcs_context_get(c);
393  if (context) {
394  for (list = context->outq_head.next;
395  list != &context->outq_head; list = list_next) {
396 
397  list_next = list->next;
398  outq_item = list_entry (list, struct outq_item, list);
399 
400  list_del (list);
401  free (outq_item->msg);
402  free (outq_item);
403  }
404  free(context);
405  }
406 }
407 
408 static int32_t cs_ipcs_connection_closed (qb_ipcs_connection_t *c)
409 {
410  int32_t res = 0;
411  int32_t service = qb_ipcs_service_id_get(c);
412  icmap_iter_t iter;
413  char prefix[ICMAP_KEYNAME_MAXLEN];
414  const char *key_name;
415  struct cs_ipcs_conn_context *cnx;
416 
417  log_printf(LOG_DEBUG, "%s() ", __func__);
418  res = corosync_service[service]->lib_exit_fn(c);
419  if (res != 0) {
420  return res;
421  }
422 
423  qb_loop_job_del(cs_poll_handle_get(), QB_LOOP_HIGH, c, outq_flush);
424 
425  cnx = qb_ipcs_context_get(c);
426 
427  snprintf(prefix, ICMAP_KEYNAME_MAXLEN, "%s.", cnx->icmap_path);
428  iter = icmap_iter_init(prefix);
429  while ((key_name = icmap_iter_next(iter, NULL, NULL)) != NULL) {
430  icmap_delete(key_name);
431  }
432  icmap_iter_finalize(iter);
433  free(cnx->icmap_path);
434 
435  icmap_inc("runtime.connections.closed");
436  icmap_dec("runtime.connections.active");
437 
438  return 0;
439 }
440 
442  const struct iovec *iov,
443  unsigned int iov_len)
444 {
445  int32_t rc = qb_ipcs_response_sendv(conn, iov, iov_len);
446  if (rc >= 0) {
447  return 0;
448  }
449  return rc;
450 }
451 
452 int cs_ipcs_response_send(void *conn, const void *msg, size_t mlen)
453 {
454  int32_t rc = qb_ipcs_response_send(conn, msg, mlen);
455  if (rc >= 0) {
456  return 0;
457  }
458  return rc;
459 }
460 
461 static void outq_flush (void *data)
462 {
463  qb_ipcs_connection_t *conn = data;
464  struct list_head *list, *list_next;
465  struct outq_item *outq_item;
466  int32_t rc;
467  struct cs_ipcs_conn_context *context = qb_ipcs_context_get(conn);
468 
469  for (list = context->outq_head.next;
470  list != &context->outq_head; list = list_next) {
471 
472  list_next = list->next;
473  outq_item = list_entry (list, struct outq_item, list);
474 
475  rc = qb_ipcs_event_send(conn, outq_item->msg, outq_item->mlen);
476  if (rc < 0 && rc != -EAGAIN) {
477  errno = -rc;
478  qb_perror(LOG_ERR, "qb_ipcs_event_send");
479  return;
480  } else if (rc == -EAGAIN) {
481  break;
482  }
483  assert(rc == outq_item->mlen);
484  context->sent++;
485  context->queued--;
486 
487  list_del (list);
488  free (outq_item->msg);
489  free (outq_item);
490  }
491  if (list_empty (&context->outq_head)) {
492  context->queuing = QB_FALSE;
493  log_printf(LOGSYS_LEVEL_INFO, "Q empty, queued:%d sent:%d.",
494  context->queued, context->sent);
495  context->queued = 0;
496  context->sent = 0;
497  } else {
498  qb_loop_job_add(cs_poll_handle_get(), QB_LOOP_HIGH, conn, outq_flush);
499  }
500 }
501 
502 static void msg_send_or_queue(qb_ipcs_connection_t *conn, const struct iovec *iov, uint32_t iov_len)
503 {
504  int32_t rc = 0;
505  int32_t i;
506  int32_t bytes_msg = 0;
507  struct outq_item *outq_item;
508  char *write_buf = 0;
509  struct cs_ipcs_conn_context *context = qb_ipcs_context_get(conn);
510 
511  for (i = 0; i < iov_len; i++) {
512  bytes_msg += iov[i].iov_len;
513  }
514 
515  if (!context->queuing) {
516  assert(list_empty (&context->outq_head));
517  rc = qb_ipcs_event_sendv(conn, iov, iov_len);
518  if (rc == bytes_msg) {
519  context->sent++;
520  return;
521  }
522  if (rc == -EAGAIN) {
523  context->queued = 0;
524  context->sent = 0;
525  context->queuing = QB_TRUE;
526  qb_loop_job_add(cs_poll_handle_get(), QB_LOOP_HIGH, conn, outq_flush);
527  } else {
528  log_printf(LOGSYS_LEVEL_ERROR, "event_send retuned %d, expected %d!", rc, bytes_msg);
529  return;
530  }
531  }
532  outq_item = malloc (sizeof (struct outq_item));
533  if (outq_item == NULL) {
534  qb_ipcs_disconnect(conn);
535  return;
536  }
537  outq_item->msg = malloc (bytes_msg);
538  if (outq_item->msg == NULL) {
539  free (outq_item);
540  qb_ipcs_disconnect(conn);
541  return;
542  }
543 
544  write_buf = outq_item->msg;
545  for (i = 0; i < iov_len; i++) {
546  memcpy (write_buf, iov[i].iov_base, iov[i].iov_len);
547  write_buf += iov[i].iov_len;
548  }
549  outq_item->mlen = bytes_msg;
550  list_init (&outq_item->list);
551  list_add_tail (&outq_item->list, &context->outq_head);
552  context->queued++;
553 }
554 
555 int cs_ipcs_dispatch_send(void *conn, const void *msg, size_t mlen)
556 {
557  struct iovec iov;
558  iov.iov_base = (void *)msg;
559  iov.iov_len = mlen;
560  msg_send_or_queue (conn, &iov, 1);
561  return 0;
562 }
563 
565  const struct iovec *iov,
566  unsigned int iov_len)
567 {
568  msg_send_or_queue(conn, iov, iov_len);
569  return 0;
570 }
571 
572 static int32_t cs_ipcs_msg_process(qb_ipcs_connection_t *c,
573  void *data, size_t size)
574 {
575  struct qb_ipc_response_header response;
576  struct qb_ipc_request_header *request_pt = (struct qb_ipc_request_header *)data;
577  int32_t service = qb_ipcs_service_id_get(c);
578  int32_t send_ok = 0;
579  int32_t is_async_call = QB_FALSE;
580  ssize_t res = -1;
581  int sending_allowed_private_data;
582  struct cs_ipcs_conn_context *cnx;
583 
584  send_ok = corosync_sending_allowed (service,
585  request_pt->id,
586  request_pt,
587  &sending_allowed_private_data);
588 
589  is_async_call = (service == CPG_SERVICE && request_pt->id == 2);
590 
591  /*
592  * This happens when the message contains some kind of invalid
593  * parameter, such as an invalid size
594  */
595  if (send_ok == -EINVAL) {
596  response.size = sizeof (response);
597  response.id = 0;
598  response.error = CS_ERR_INVALID_PARAM;
599 
600  cnx = qb_ipcs_context_get(c);
601  if (cnx) {
602  cnx->invalid_request++;
603  }
604 
605  if (is_async_call) {
606  log_printf(LOGSYS_LEVEL_INFO, "*** %s() invalid message! size:%d error:%d",
607  __func__, response.size, response.error);
608  } else {
609  qb_ipcs_response_send (c,
610  &response,
611  sizeof (response));
612  }
613  res = -EINVAL;
614  } else if (send_ok < 0) {
615  cnx = qb_ipcs_context_get(c);
616  if (cnx) {
617  cnx->overload++;
618  }
619  if (!is_async_call) {
620  /*
621  * Overload, tell library to retry
622  */
623  response.size = sizeof (response);
624  response.id = 0;
625  response.error = CS_ERR_TRY_AGAIN;
626  qb_ipcs_response_send (c,
627  &response,
628  sizeof (response));
629  } else {
631  "*** %s() (%d:%d - %d) %s!",
632  __func__, service, request_pt->id,
633  is_async_call, strerror(-send_ok));
634  }
635  res = -ENOBUFS;
636  }
637 
638  if (send_ok >= 0) {
639  corosync_service[service]->lib_engine[request_pt->id].lib_handler_fn(c, request_pt);
640  res = 0;
641  }
642  corosync_sending_allowed_release (&sending_allowed_private_data);
643  return res;
644 }
645 
646 
647 static int32_t cs_ipcs_job_add(enum qb_loop_priority p, void *data, qb_loop_job_dispatch_fn fn)
648 {
649  return qb_loop_job_add(cs_poll_handle_get(), p, data, fn);
650 }
651 
652 static int32_t cs_ipcs_dispatch_add(enum qb_loop_priority p, int32_t fd, int32_t events,
653  void *data, qb_ipcs_dispatch_fn_t fn)
654 {
655  return qb_loop_poll_add(cs_poll_handle_get(), p, fd, events, data, fn);
656 }
657 
658 static int32_t cs_ipcs_dispatch_mod(enum qb_loop_priority p, int32_t fd, int32_t events,
659  void *data, qb_ipcs_dispatch_fn_t fn)
660 {
661  return qb_loop_poll_mod(cs_poll_handle_get(), p, fd, events, data, fn);
662 }
663 
664 static int32_t cs_ipcs_dispatch_del(int32_t fd)
665 {
666  return qb_loop_poll_del(cs_poll_handle_get(), fd);
667 }
668 
669 static void cs_ipcs_low_fds_event(int32_t not_enough, int32_t fds_available)
670 {
671  ipc_not_enough_fds_left = not_enough;
672  if (not_enough) {
673  log_printf(LOGSYS_LEVEL_WARNING, "refusing new connections (fds_available:%d)",
674  fds_available);
675  } else {
676  log_printf(LOGSYS_LEVEL_NOTICE, "allowing new connections (fds_available:%d)",
677  fds_available);
678 
679  }
680 }
681 
682 int32_t cs_ipcs_q_level_get(void)
683 {
684  return ipc_fc_totem_queue_level;
685 }
686 
687 static qb_loop_timer_handle ipcs_check_for_flow_control_timer;
688 static void cs_ipcs_check_for_flow_control(void)
689 {
690  int32_t i;
691  int32_t fc_enabled;
692 
693  for (i = 0; i < SERVICES_COUNT_MAX; i++) {
694  if (corosync_service[i] == NULL || ipcs_mapper[i].inst == NULL) {
695  continue;
696  }
697  fc_enabled = QB_IPCS_RATE_OFF;
698  if (ipc_fc_is_quorate == 1 ||
699  corosync_service[i]->allow_inquorate == CS_LIB_ALLOW_INQUORATE) {
700  /*
701  * we are quorate
702  * now check flow control
703  */
704  if (ipc_fc_totem_queue_level != TOTEM_Q_LEVEL_CRITICAL &&
705  ipc_fc_sync_in_process == 0) {
706  fc_enabled = QB_FALSE;
707  } else if (ipc_fc_totem_queue_level != TOTEM_Q_LEVEL_CRITICAL &&
708  i == VOTEQUORUM_SERVICE) {
709  /*
710  * Allow message processing for votequorum service even
711  * in sync phase
712  */
713  fc_enabled = QB_FALSE;
714  } else {
715  fc_enabled = QB_IPCS_RATE_OFF_2;
716  }
717  }
718  if (fc_enabled) {
719  qb_ipcs_request_rate_limit(ipcs_mapper[i].inst, fc_enabled);
720 
721  qb_loop_timer_add(cs_poll_handle_get(), QB_LOOP_MED, 1*QB_TIME_NS_IN_MSEC,
722  NULL, corosync_recheck_the_q_level, &ipcs_check_for_flow_control_timer);
723  } else if (ipc_fc_totem_queue_level == TOTEM_Q_LEVEL_LOW) {
724  qb_ipcs_request_rate_limit(ipcs_mapper[i].inst, QB_IPCS_RATE_FAST);
725  } else if (ipc_fc_totem_queue_level == TOTEM_Q_LEVEL_GOOD) {
726  qb_ipcs_request_rate_limit(ipcs_mapper[i].inst, QB_IPCS_RATE_NORMAL);
727  } else if (ipc_fc_totem_queue_level == TOTEM_Q_LEVEL_HIGH) {
728  qb_ipcs_request_rate_limit(ipcs_mapper[i].inst, QB_IPCS_RATE_SLOW);
729  }
730  }
731 }
732 
733 static void cs_ipcs_fc_quorum_changed(int quorate, void *context)
734 {
735  ipc_fc_is_quorate = quorate;
736  cs_ipcs_check_for_flow_control();
737 }
738 
739 static void cs_ipcs_totem_queue_level_changed(enum totem_q_level level)
740 {
741  ipc_fc_totem_queue_level = level;
742  cs_ipcs_check_for_flow_control();
743 }
744 
745 void cs_ipcs_sync_state_changed(int32_t sync_in_process)
746 {
747  ipc_fc_sync_in_process = sync_in_process;
748  cs_ipcs_check_for_flow_control();
749 }
750 
752 {
753  int32_t i;
754  struct qb_ipcs_stats srv_stats;
755  struct qb_ipcs_connection_stats stats;
756  qb_ipcs_connection_t *c, *prev;
757  struct cs_ipcs_conn_context *cnx;
758  char key_name[ICMAP_KEYNAME_MAXLEN];
759 
760  for (i = 0; i < SERVICES_COUNT_MAX; i++) {
761  if (corosync_service[i] == NULL || ipcs_mapper[i].inst == NULL) {
762  continue;
763  }
764  qb_ipcs_stats_get(ipcs_mapper[i].inst, &srv_stats, QB_FALSE);
765 
766  for (c = qb_ipcs_connection_first_get(ipcs_mapper[i].inst);
767  c;
768  prev = c, c = qb_ipcs_connection_next_get(ipcs_mapper[i].inst, prev), qb_ipcs_connection_unref(prev)) {
769 
770  cnx = qb_ipcs_context_get(c);
771  if (cnx == NULL) continue;
772 
773  qb_ipcs_connection_stats_get(c, &stats, QB_FALSE);
774 
775  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.client_pid", cnx->icmap_path);
776  icmap_set_uint32(key_name, stats.client_pid);
777 
778  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.requests", cnx->icmap_path);
779  icmap_set_uint64(key_name, stats.requests);
780 
781  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.responses", cnx->icmap_path);
782  icmap_set_uint64(key_name, stats.responses);
783 
784  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.dispatched", cnx->icmap_path);
785  icmap_set_uint64(key_name, stats.events);
786 
787  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.send_retries", cnx->icmap_path);
788  icmap_set_uint64(key_name, stats.send_retries);
789 
790  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.recv_retries", cnx->icmap_path);
791  icmap_set_uint64(key_name, stats.recv_retries);
792 
793  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.flow_control", cnx->icmap_path);
794  icmap_set_uint32(key_name, stats.flow_control_state);
795 
796  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.flow_control_count", cnx->icmap_path);
797  icmap_set_uint64(key_name, stats.flow_control_count);
798 
799  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.queue_size", cnx->icmap_path);
800  icmap_set_uint32(key_name, cnx->queued);
801 
802  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.invalid_request", cnx->icmap_path);
803  icmap_set_uint64(key_name, cnx->invalid_request);
804 
805  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "%s.overload", cnx->icmap_path);
806  icmap_set_uint64(key_name, cnx->overload);
807  }
808  }
809 }
810 
811 static enum qb_ipc_type cs_get_ipc_type (void)
812 {
813  char *str;
814  int found = 0;
815  enum qb_ipc_type ret = QB_IPC_NATIVE;
816 
817  if (icmap_get_string("qb.ipc_type", &str) != CS_OK) {
818  log_printf(LOGSYS_LEVEL_DEBUG, "No configured qb.ipc_type. Using native ipc");
819  return QB_IPC_NATIVE;
820  }
821 
822  if (strcmp(str, "native") == 0) {
823  ret = QB_IPC_NATIVE;
824  found = 1;
825  }
826 
827  if (strcmp(str, "shm") == 0) {
828  ret = QB_IPC_SHM;
829  found = 1;
830  }
831 
832  if (strcmp(str, "socket") == 0) {
833  ret = QB_IPC_SOCKET;
834  found = 1;
835  }
836 
837  if (found) {
838  log_printf(LOGSYS_LEVEL_DEBUG, "Using %s ipc", str);
839  } else {
840  log_printf(LOGSYS_LEVEL_DEBUG, "Unknown ipc type %s", str);
841  }
842 
843  free(str);
844 
845  return ret;
846 }
847 
848 const char *cs_ipcs_service_init(struct corosync_service_engine *service)
849 {
850  const char *serv_short_name;
851 
852  serv_short_name = cs_ipcs_serv_short_name(service->id);
853 
854  if (service->lib_engine_count == 0) {
856  "NOT Initializing IPC on %s [%d]",
857  serv_short_name,
858  service->id);
859  return NULL;
860  }
861 
862  if (strlen(serv_short_name) >= CS_IPCS_MAPPER_SERV_NAME) {
863  log_printf (LOGSYS_LEVEL_ERROR, "service name %s is too long", serv_short_name);
864  return "qb_ipcs_run error";
865  }
866 
867  ipcs_mapper[service->id].id = service->id;
868  strcpy(ipcs_mapper[service->id].name, serv_short_name);
870  "Initializing IPC on %s [%d]",
871  ipcs_mapper[service->id].name,
872  ipcs_mapper[service->id].id);
873  ipcs_mapper[service->id].inst = qb_ipcs_create(ipcs_mapper[service->id].name,
874  ipcs_mapper[service->id].id,
875  cs_get_ipc_type(),
876  &corosync_service_funcs);
877  assert(ipcs_mapper[service->id].inst);
878  qb_ipcs_poll_handlers_set(ipcs_mapper[service->id].inst,
879  &corosync_poll_funcs);
880  if (qb_ipcs_run(ipcs_mapper[service->id].inst) != 0) {
881  log_printf (LOGSYS_LEVEL_ERROR, "Can't initialize IPC");
882  return "qb_ipcs_run error";
883  }
884 
885  return NULL;
886 }
887 
888 void cs_ipcs_init(void)
889 {
890  api = apidef_get ();
891 
892  qb_loop_poll_low_fds_event_set(cs_poll_handle_get(), cs_ipcs_low_fds_event);
893 
894  api->quorum_register_callback (cs_ipcs_fc_quorum_changed, NULL);
895  totempg_queue_level_register_callback (cs_ipcs_totem_queue_level_changed);
896 
897  icmap_set_uint64("runtime.connections.active", 0);
898  icmap_set_uint64("runtime.connections.closed", 0);
899 }
900 
void cs_ipc_refcnt_dec(void *conn)
Definition: ipc_glue.c:372
int32_t cs_ipcs_q_level_get(void)
Definition: ipc_glue.c:682
const char * cs_ipcs_service_init(struct corosync_service_engine *service)
Definition: ipc_glue.c:848
#define CS_IPCS_MAPPER_SERV_NAME
Definition: ipc_glue.c:73
const char * icmap_iter_next(icmap_iter_t iter, size_t *value_len, icmap_value_types_t *type)
Definition: icmap.c:1103
void * msg
Definition: ipc_glue.c:82
#define LOGSYS_LEVEL_INFO
Definition: logsys.h:73
struct list_head list
Definition: ipc_glue.c:84
void cs_ipc_refcnt_inc(void *conn)
Definition: ipc_glue.c:367
struct list_head * next
Definition: list.h:47
struct list_head outq_head
Definition: ipc_glue.c:252
Totem Single Ring Protocol.
void icmap_iter_finalize(icmap_iter_t iter)
Definition: icmap.c:1124
qb_loop_t * cs_poll_handle_get(void)
Definition: main.c:163
void corosync_recheck_the_q_level(void *data)
Definition: main.c:752
size_t mlen
Definition: ipc_glue.c:83
int(* quorum_register_callback)(quorum_callback_fn_t callback_fn, void *context)
Definition: coroapi.h:339
qb_ipcs_service_t * inst
Definition: ipc_glue.c:77
cs_error_t icmap_set_string(const char *key_name, const char *value)
Definition: icmap.c:641
int cs_ipcs_dispatch_iov_send(void *conn, const struct iovec *iov, unsigned int iov_len)
Definition: ipc_glue.c:564
struct corosync_service_engine * corosync_service[SERVICES_COUNT_MAX]
Definition: service.c:110
Definition: list.h:46
cs_error_t icmap_inc(const char *key_name)
Definition: icmap.c:1057
#define log_printf(level, format, args...)
Definition: logsys.h:217
void cs_ipcs_sync_state_changed(int32_t sync_in_process)
Definition: ipc_glue.c:745
int corosync_sending_allowed(unsigned int service, unsigned int id, const void *msg, void *sending_allowed_private_data)
Definition: main.c:766
int cs_ipcs_response_send(void *conn, const void *msg, size_t mlen)
Definition: ipc_glue.c:452
void cs_ipc_allow_connections(int32_t allow)
Definition: ipc_glue.c:155
int(* lib_exit_fn)(void *conn)
Definition: coroapi.h:444
#define ICMAP_KEYNAME_MAXLEN
Definition: icmap.h:48
cs_error_t icmap_get_uint8(const char *key_name, uint8_t *u8)
Definition: icmap.c:842
uint64_t invalid_request
Definition: ipc_glue.c:255
#define LOGSYS_LEVEL_WARNING
Definition: logsys.h:71
struct corosync_lib_handler * lib_engine
Definition: coroapi.h:445
cs_error_t icmap_set_uint32(const char *key_name, uint32_t value)
Definition: icmap.c:611
#define LOGSYS_LEVEL_ERROR
Definition: logsys.h:70
cs_error_t icmap_delete(const char *key_name)
Definition: icmap.c:667
int cs_ipcs_response_iov_send(void *conn, const struct iovec *iov, unsigned int iov_len)
Definition: ipc_glue.c:441
#define LOGSYS_LEVEL_DEBUG
Definition: logsys.h:74
void cs_ipcs_init(void)
Definition: ipc_glue.c:888
LOGSYS_DECLARE_SUBSYS("MAIN")
cs_error_t icmap_dec(const char *key_name)
Definition: icmap.c:1067
uint32_t quorate
Definition: sam.c:133
cs_error_t icmap_set_uint64(const char *key_name, uint64_t value)
Definition: icmap.c:623
struct corosync_api_v1 * apidef_get(void)
Definition: apidef.c:147
void corosync_sending_allowed_release(void *sending_allowed_private_data)
Definition: main.c:809
int32_t id
Definition: ipc_glue.c:76
void cs_ipcs_stats_update(void)
Definition: ipc_glue.c:751
unsigned short id
Definition: coroapi.h:433
#define SERVICES_COUNT_MAX
Definition: coroapi.h:415
int32_t cs_ipcs_service_destroy(int32_t service_id)
Definition: ipc_glue.c:160
void totempg_queue_level_register_callback(totem_queue_level_changed_fn)
Definition: totempg.c:1517
cs_error_t icmap_get_string(const char *key_name, char **str)
Definition: icmap.c:896
#define list_entry(ptr, type, member)
Definition: list.h:84
char name[CS_IPCS_MAPPER_SERV_NAME]
Definition: ipc_glue.c:78
void * cs_ipcs_private_data_get(void *conn)
Definition: ipc_glue.c:377
int cs_ipcs_dispatch_send(void *conn, const void *msg, size_t mlen)
Definition: ipc_glue.c:555
#define LOGSYS_LEVEL_NOTICE
Definition: logsys.h:72
void(* lib_handler_fn)(void *conn, const void *msg)
Definition: coroapi.h:418
totem_q_level
Definition: totempg.h:173
void icmap_convert_name_to_valid_name(char *key_name)
Definition: icmap.c:292
unsigned int private_data_size
Definition: coroapi.h:437
icmap_iter_t icmap_iter_init(const char *prefix)
Definition: icmap.c:1097
qb_map_iter_t * icmap_iter_t
Definition: icmap.h:121