corosync  3.0.2
ipc_glue.c
Go to the documentation of this file.
1 /*
2  * Copyright (c) 2010-2017 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 #include "ipcs_stats.h"
64 #include "stats.h"
65 
66 LOGSYS_DECLARE_SUBSYS ("MAIN");
67 
68 static struct corosync_api_v1 *api = NULL;
69 static int32_t ipc_not_enough_fds_left = 0;
70 static int32_t ipc_fc_is_quorate; /* boolean */
71 static int32_t ipc_fc_totem_queue_level; /* percentage used */
72 static int32_t ipc_fc_sync_in_process; /* boolean */
73 static int32_t ipc_allow_connections = 0; /* boolean */
74 
75 #define CS_IPCS_MAPPER_SERV_NAME 256
76 
78  int32_t id;
79  qb_ipcs_service_t *inst;
81 };
82 
83 struct outq_item {
84  void *msg;
85  size_t mlen;
86  struct qb_list_head list;
87 };
88 
89 static struct cs_ipcs_mapper ipcs_mapper[SERVICES_COUNT_MAX];
90 
91 static int32_t cs_ipcs_job_add(enum qb_loop_priority p, void *data, qb_loop_job_dispatch_fn fn);
92 static int32_t cs_ipcs_dispatch_add(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_mod(enum qb_loop_priority p, int32_t fd, int32_t events,
95  void *data, qb_ipcs_dispatch_fn_t fn);
96 static int32_t cs_ipcs_dispatch_del(int32_t fd);
97 static void outq_flush (void *data);
98 
99 
100 static struct qb_ipcs_poll_handlers corosync_poll_funcs = {
101  .job_add = cs_ipcs_job_add,
102  .dispatch_add = cs_ipcs_dispatch_add,
103  .dispatch_mod = cs_ipcs_dispatch_mod,
104  .dispatch_del = cs_ipcs_dispatch_del,
105 };
106 
107 static int32_t cs_ipcs_connection_accept (qb_ipcs_connection_t *c, uid_t euid, gid_t egid);
108 static void cs_ipcs_connection_created(qb_ipcs_connection_t *c);
109 static int32_t cs_ipcs_msg_process(qb_ipcs_connection_t *c,
110  void *data, size_t size);
111 static int32_t cs_ipcs_connection_closed (qb_ipcs_connection_t *c);
112 static void cs_ipcs_connection_destroyed (qb_ipcs_connection_t *c);
113 
114 static struct qb_ipcs_service_handlers corosync_service_funcs = {
115  .connection_accept = cs_ipcs_connection_accept,
116  .connection_created = cs_ipcs_connection_created,
117  .msg_process = cs_ipcs_msg_process,
118  .connection_closed = cs_ipcs_connection_closed,
119  .connection_destroyed = cs_ipcs_connection_destroyed,
120 };
121 
122 static struct ipcs_global_stats global_stats;
123 
124 static const char* cs_ipcs_serv_short_name(int32_t service_id)
125 {
126  const char *name;
127  switch (service_id) {
128  case CFG_SERVICE:
129  name = "cfg";
130  break;
131  case CPG_SERVICE:
132  name = "cpg";
133  break;
134  case QUORUM_SERVICE:
135  name = "quorum";
136  break;
137  case PLOAD_SERVICE:
138  name = "pload";
139  break;
140  case VOTEQUORUM_SERVICE:
141  name = "votequorum";
142  break;
143  case MON_SERVICE:
144  name = "mon";
145  break;
146  case WD_SERVICE:
147  name = "wd";
148  break;
149  case CMAP_SERVICE:
150  name = "cmap";
151  break;
152  default:
153  name = NULL;
154  break;
155  }
156  return name;
157 }
158 
159 void cs_ipc_allow_connections(int32_t allow)
160 {
161  ipc_allow_connections = allow;
162 }
163 
164 int32_t cs_ipcs_service_destroy(int32_t service_id)
165 {
166  if (ipcs_mapper[service_id].inst) {
167  qb_ipcs_destroy(ipcs_mapper[service_id].inst);
168  ipcs_mapper[service_id].inst = NULL;
169  }
170  return 0;
171 }
172 
173 static int32_t cs_ipcs_connection_accept (qb_ipcs_connection_t *c, uid_t euid, gid_t egid)
174 {
175  int32_t service = qb_ipcs_service_id_get(c);
176  uint8_t u8;
177  char key_name[ICMAP_KEYNAME_MAXLEN];
178 
179  if (!ipc_allow_connections) {
180  log_printf(LOGSYS_LEVEL_DEBUG, "Denied connection, corosync is not ready");
181  return -EAGAIN;
182  }
183 
184  if (corosync_service[service] == NULL ||
185  ipcs_mapper[service].inst == NULL) {
186  return -ENOSYS;
187  }
188 
189  if (ipc_not_enough_fds_left) {
190  return -EMFILE;
191  }
192 
193  if (euid == 0 || egid == 0) {
194  return 0;
195  }
196 
197  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "uidgid.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.config.uid.%u", euid);
202  if (icmap_get_uint8(key_name, &u8) == CS_OK && u8 == 1)
203  return 0;
204 
205  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "uidgid.gid.%u", egid);
206  if (icmap_get_uint8(key_name, &u8) == CS_OK && u8 == 1)
207  return 0;
208 
209  snprintf(key_name, ICMAP_KEYNAME_MAXLEN, "uidgid.config.gid.%u", egid);
210  if (icmap_get_uint8(key_name, &u8) == CS_OK && u8 == 1)
211  return 0;
212 
213  log_printf(LOGSYS_LEVEL_ERROR, "Denied connection attempt from %d:%d", euid, egid);
214 
215  return -EACCES;
216 }
217 
218 static char * pid_to_name (pid_t pid, char *out_name, size_t name_len)
219 {
220  char *name;
221  char *rest;
222  FILE *fp;
223  char fname[32];
224  char buf[256];
225 
226  snprintf (fname, 32, "/proc/%d/stat", pid);
227  fp = fopen (fname, "r");
228  if (!fp) {
229  return NULL;
230  }
231 
232  if (fgets (buf, sizeof (buf), fp) == NULL) {
233  fclose (fp);
234  return NULL;
235  }
236  fclose (fp);
237 
238  name = strrchr (buf, '(');
239  if (!name) {
240  return NULL;
241  }
242 
243  /* move past the bracket */
244  name++;
245 
246  rest = strrchr (buf, ')');
247 
248  if (rest == NULL || rest[1] != ' ') {
249  return NULL;
250  }
251 
252  *rest = '\0';
253  /* move past the NULL and space */
254  rest += 2;
255 
256  /* copy the name */
257  strncpy (out_name, name, name_len - 1);
258  out_name[name_len - 1] = '\0';
259  return out_name;
260 }
261 
262 static void cs_ipcs_connection_created(qb_ipcs_connection_t *c)
263 {
264  int32_t service = 0;
265  struct cs_ipcs_conn_context *context;
266  struct qb_ipcs_connection_stats stats;
267  size_t size = sizeof(struct cs_ipcs_conn_context);
268 
269  log_printf(LOG_DEBUG, "connection created");
270 
271  service = qb_ipcs_service_id_get(c);
272 
273  size += corosync_service[service]->private_data_size;
274  context = calloc(1, size);
275  if (context == NULL) {
276  qb_ipcs_disconnect(c);
277  return;
278  }
279 
280  qb_list_init(&context->outq_head);
281  context->queuing = QB_FALSE;
282  context->queued = 0;
283  context->sent = 0;
284 
285  qb_ipcs_context_set(c, context);
286 
287  if (corosync_service[service]->lib_init_fn(c) != 0) {
288  log_printf(LOG_ERR, "lib_init_fn failed, disconnecting");
289  qb_ipcs_disconnect(c);
290  return;
291  }
292 
293  qb_ipcs_connection_stats_get(c, &stats, QB_FALSE);
294 
295  if (!pid_to_name (stats.client_pid, context->proc_name, sizeof(context->proc_name))) {
296  context->proc_name[0] = '\0';
297  }
298  stats_ipcs_add_connection(service, stats.client_pid, c);
299  global_stats.active++;
300 }
301 
302 void cs_ipc_refcnt_inc(void *conn)
303 {
304  qb_ipcs_connection_ref(conn);
305 }
306 
307 void cs_ipc_refcnt_dec(void *conn)
308 {
309  qb_ipcs_connection_unref(conn);
310 }
311 
312 void *cs_ipcs_private_data_get(void *conn)
313 {
314  struct cs_ipcs_conn_context *cnx;
315  cnx = qb_ipcs_context_get(conn);
316  return &cnx->data[0];
317 }
318 
319 static void cs_ipcs_connection_destroyed (qb_ipcs_connection_t *c)
320 {
321  struct cs_ipcs_conn_context *context;
322  struct qb_list_head *list, *tmp_iter;
323  struct outq_item *outq_item;
324 
325  log_printf(LOG_DEBUG, "%s() ", __func__);
326 
327  context = qb_ipcs_context_get(c);
328  if (context) {
329  qb_list_for_each_safe(list, tmp_iter, &(context->outq_head)) {
330  outq_item = qb_list_entry (list, struct outq_item, list);
331 
332  qb_list_del (list);
333  free (outq_item->msg);
334  free (outq_item);
335  }
336  free(context);
337  }
338 }
339 
340 static int32_t cs_ipcs_connection_closed (qb_ipcs_connection_t *c)
341 {
342  int32_t res = 0;
343  int32_t service = qb_ipcs_service_id_get(c);
344  struct qb_ipcs_connection_stats stats;
345 
346  log_printf(LOG_DEBUG, "%s() ", __func__);
347  res = corosync_service[service]->lib_exit_fn(c);
348  if (res != 0) {
349  return res;
350  }
351 
352  qb_loop_job_del(cs_poll_handle_get(), QB_LOOP_HIGH, c, outq_flush);
353 
354  qb_ipcs_connection_stats_get(c, &stats, QB_FALSE);
355 
356  stats_ipcs_del_connection(service, stats.client_pid, c);
357 
358  global_stats.active--;
359  global_stats.closed++;
360  return 0;
361 }
362 
364  const struct iovec *iov,
365  unsigned int iov_len)
366 {
367  int32_t rc = qb_ipcs_response_sendv(conn, iov, iov_len);
368  if (rc >= 0) {
369  return 0;
370  }
371  return rc;
372 }
373 
374 int cs_ipcs_response_send(void *conn, const void *msg, size_t mlen)
375 {
376  int32_t rc = qb_ipcs_response_send(conn, msg, mlen);
377  if (rc >= 0) {
378  return 0;
379  }
380  return rc;
381 }
382 
383 static void outq_flush (void *data)
384 {
385  qb_ipcs_connection_t *conn = data;
386  struct qb_list_head *list, *tmp_iter;
387  struct outq_item *outq_item;
388  int32_t rc;
389  struct cs_ipcs_conn_context *context = qb_ipcs_context_get(conn);
390 
391  qb_list_for_each_safe(list, tmp_iter, &(context->outq_head)) {
392  outq_item = qb_list_entry (list, struct outq_item, list);
393 
394  rc = qb_ipcs_event_send(conn, outq_item->msg, outq_item->mlen);
395  if (rc < 0 && rc != -EAGAIN) {
396  errno = -rc;
397  qb_perror(LOG_ERR, "qb_ipcs_event_send");
398  return;
399  } else if (rc == -EAGAIN) {
400  break;
401  }
402  assert(rc == outq_item->mlen);
403  context->sent++;
404  context->queued--;
405 
406  qb_list_del (list);
407  free (outq_item->msg);
408  free (outq_item);
409  }
410  if (qb_list_empty (&context->outq_head)) {
411  context->queuing = QB_FALSE;
412  log_printf(LOGSYS_LEVEL_INFO, "Q empty, queued:%d sent:%d.",
413  context->queued, context->sent);
414  context->queued = 0;
415  context->sent = 0;
416  } else {
417  qb_loop_job_add(cs_poll_handle_get(), QB_LOOP_HIGH, conn, outq_flush);
418  }
419 }
420 
421 static void msg_send_or_queue(qb_ipcs_connection_t *conn, const struct iovec *iov, uint32_t iov_len)
422 {
423  int32_t rc = 0;
424  int32_t i;
425  int32_t bytes_msg = 0;
426  struct outq_item *outq_item;
427  char *write_buf = 0;
428  struct cs_ipcs_conn_context *context = qb_ipcs_context_get(conn);
429 
430  for (i = 0; i < iov_len; i++) {
431  bytes_msg += iov[i].iov_len;
432  }
433 
434  if (!context->queuing) {
435  assert(qb_list_empty (&context->outq_head));
436  rc = qb_ipcs_event_sendv(conn, iov, iov_len);
437  if (rc == bytes_msg) {
438  context->sent++;
439  return;
440  }
441  if (rc == -EAGAIN) {
442  context->queued = 0;
443  context->sent = 0;
444  context->queuing = QB_TRUE;
445  qb_loop_job_add(cs_poll_handle_get(), QB_LOOP_HIGH, conn, outq_flush);
446  } else {
447  log_printf(LOGSYS_LEVEL_ERROR, "event_send retuned %d, expected %d!", rc, bytes_msg);
448  return;
449  }
450  }
451  outq_item = malloc (sizeof (struct outq_item));
452  if (outq_item == NULL) {
453  qb_ipcs_disconnect(conn);
454  return;
455  }
456  outq_item->msg = malloc (bytes_msg);
457  if (outq_item->msg == NULL) {
458  free (outq_item);
459  qb_ipcs_disconnect(conn);
460  return;
461  }
462 
463  write_buf = outq_item->msg;
464  for (i = 0; i < iov_len; i++) {
465  memcpy (write_buf, iov[i].iov_base, iov[i].iov_len);
466  write_buf += iov[i].iov_len;
467  }
468  outq_item->mlen = bytes_msg;
469  qb_list_init (&outq_item->list);
470  qb_list_add_tail (&outq_item->list, &context->outq_head);
471  context->queued++;
472 }
473 
474 int cs_ipcs_dispatch_send(void *conn, const void *msg, size_t mlen)
475 {
476  struct iovec iov;
477  iov.iov_base = (void *)msg;
478  iov.iov_len = mlen;
479  msg_send_or_queue (conn, &iov, 1);
480  return 0;
481 }
482 
484  const struct iovec *iov,
485  unsigned int iov_len)
486 {
487  msg_send_or_queue(conn, iov, iov_len);
488  return 0;
489 }
490 
491 static int32_t cs_ipcs_msg_process(qb_ipcs_connection_t *c,
492  void *data, size_t size)
493 {
494  struct qb_ipc_response_header response;
495  struct qb_ipc_request_header *request_pt = (struct qb_ipc_request_header *)data;
496  int32_t service = qb_ipcs_service_id_get(c);
497  int32_t send_ok = 0;
498  int32_t is_async_call = QB_FALSE;
499  ssize_t res = -1;
500  int sending_allowed_private_data;
501  struct cs_ipcs_conn_context *cnx;
502 
503  send_ok = corosync_sending_allowed (service,
504  request_pt->id,
505  request_pt,
506  &sending_allowed_private_data);
507 
508  is_async_call = (service == CPG_SERVICE && request_pt->id == 2);
509 
510  /*
511  * This happens when the message contains some kind of invalid
512  * parameter, such as an invalid size
513  */
514  if (send_ok == -EINVAL) {
515  response.size = sizeof (response);
516  response.id = 0;
517  response.error = CS_ERR_INVALID_PARAM;
518 
519  cnx = qb_ipcs_context_get(c);
520  if (cnx) {
521  cnx->invalid_request++;
522  }
523 
524  if (is_async_call) {
525  log_printf(LOGSYS_LEVEL_INFO, "*** %s() invalid message! size:%d error:%d",
526  __func__, response.size, response.error);
527  } else {
528  qb_ipcs_response_send (c,
529  &response,
530  sizeof (response));
531  }
532  res = -EINVAL;
533  } else if (send_ok < 0) {
534  cnx = qb_ipcs_context_get(c);
535  if (cnx) {
536  cnx->overload++;
537  }
538  if (!is_async_call) {
539  /*
540  * Overload, tell library to retry
541  */
542  response.size = sizeof (response);
543  response.id = 0;
544  response.error = CS_ERR_TRY_AGAIN;
545  qb_ipcs_response_send (c,
546  &response,
547  sizeof (response));
548  } else {
550  "*** %s() (%d:%d - %d) %s!",
551  __func__, service, request_pt->id,
552  is_async_call, strerror(-send_ok));
553  }
554  res = -ENOBUFS;
555  }
556 
557  if (send_ok >= 0) {
558  corosync_service[service]->lib_engine[request_pt->id].lib_handler_fn(c, request_pt);
559  res = 0;
560  }
561  corosync_sending_allowed_release (&sending_allowed_private_data);
562  return res;
563 }
564 
565 
566 static int32_t cs_ipcs_job_add(enum qb_loop_priority p, void *data, qb_loop_job_dispatch_fn fn)
567 {
568  return qb_loop_job_add(cs_poll_handle_get(), p, data, fn);
569 }
570 
571 static int32_t cs_ipcs_dispatch_add(enum qb_loop_priority p, int32_t fd, int32_t events,
572  void *data, qb_ipcs_dispatch_fn_t fn)
573 {
574  return qb_loop_poll_add(cs_poll_handle_get(), p, fd, events, data, fn);
575 }
576 
577 static int32_t cs_ipcs_dispatch_mod(enum qb_loop_priority p, int32_t fd, int32_t events,
578  void *data, qb_ipcs_dispatch_fn_t fn)
579 {
580  return qb_loop_poll_mod(cs_poll_handle_get(), p, fd, events, data, fn);
581 }
582 
583 static int32_t cs_ipcs_dispatch_del(int32_t fd)
584 {
585  return qb_loop_poll_del(cs_poll_handle_get(), fd);
586 }
587 
588 static void cs_ipcs_low_fds_event(int32_t not_enough, int32_t fds_available)
589 {
590  ipc_not_enough_fds_left = not_enough;
591  if (not_enough) {
592  log_printf(LOGSYS_LEVEL_WARNING, "refusing new connections (fds_available:%d)",
593  fds_available);
594  } else {
595  log_printf(LOGSYS_LEVEL_NOTICE, "allowing new connections (fds_available:%d)",
596  fds_available);
597 
598  }
599 }
600 
601 int32_t cs_ipcs_q_level_get(void)
602 {
603  return ipc_fc_totem_queue_level;
604 }
605 
606 static qb_loop_timer_handle ipcs_check_for_flow_control_timer;
607 static void cs_ipcs_check_for_flow_control(void)
608 {
609  int32_t i;
610  int32_t fc_enabled;
611 
612  for (i = 0; i < SERVICES_COUNT_MAX; i++) {
613  if (corosync_service[i] == NULL || ipcs_mapper[i].inst == NULL) {
614  continue;
615  }
616  fc_enabled = QB_IPCS_RATE_OFF;
617  if (ipc_fc_is_quorate == 1 ||
618  corosync_service[i]->allow_inquorate == CS_LIB_ALLOW_INQUORATE) {
619  /*
620  * we are quorate
621  * now check flow control
622  */
623  if (ipc_fc_totem_queue_level != TOTEM_Q_LEVEL_CRITICAL &&
624  ipc_fc_sync_in_process == 0) {
625  fc_enabled = QB_FALSE;
626  } else if (ipc_fc_totem_queue_level != TOTEM_Q_LEVEL_CRITICAL &&
627  i == VOTEQUORUM_SERVICE) {
628  /*
629  * Allow message processing for votequorum service even
630  * in sync phase
631  */
632  fc_enabled = QB_FALSE;
633  } else {
634  fc_enabled = QB_IPCS_RATE_OFF_2;
635  }
636  }
637  if (fc_enabled) {
638  qb_ipcs_request_rate_limit(ipcs_mapper[i].inst, fc_enabled);
639 
640  qb_loop_timer_add(cs_poll_handle_get(), QB_LOOP_MED, 1*QB_TIME_NS_IN_MSEC,
641  NULL, corosync_recheck_the_q_level, &ipcs_check_for_flow_control_timer);
642  } else if (ipc_fc_totem_queue_level == TOTEM_Q_LEVEL_LOW) {
643  qb_ipcs_request_rate_limit(ipcs_mapper[i].inst, QB_IPCS_RATE_FAST);
644  } else if (ipc_fc_totem_queue_level == TOTEM_Q_LEVEL_GOOD) {
645  qb_ipcs_request_rate_limit(ipcs_mapper[i].inst, QB_IPCS_RATE_NORMAL);
646  } else if (ipc_fc_totem_queue_level == TOTEM_Q_LEVEL_HIGH) {
647  qb_ipcs_request_rate_limit(ipcs_mapper[i].inst, QB_IPCS_RATE_SLOW);
648  }
649  }
650 }
651 
652 static void cs_ipcs_fc_quorum_changed(int quorate, void *context)
653 {
654  ipc_fc_is_quorate = quorate;
655  cs_ipcs_check_for_flow_control();
656 }
657 
658 static void cs_ipcs_totem_queue_level_changed(enum totem_q_level level)
659 {
660  ipc_fc_totem_queue_level = level;
661  cs_ipcs_check_for_flow_control();
662 }
663 
664 void cs_ipcs_sync_state_changed(int32_t sync_in_process)
665 {
666  ipc_fc_sync_in_process = sync_in_process;
667  cs_ipcs_check_for_flow_control();
668 }
669 
671 {
672  memcpy(ipcs_stats, &global_stats, sizeof(global_stats));
673 }
674 
675 cs_error_t cs_ipcs_get_conn_stats(int service_id, uint32_t pid, void *conn_ptr, struct ipcs_conn_stats *ipcs_stats)
676 {
677  struct cs_ipcs_conn_context *cnx;
678  qb_ipcs_connection_t *c, *prev;
679  int found = 0;
680 
681  if (corosync_service[service_id] == NULL || ipcs_mapper[service_id].inst == NULL) {
682  return CS_ERR_NOT_EXIST;
683  }
684 
685  qb_ipcs_stats_get(ipcs_mapper[service_id].inst, &ipcs_stats->srv, QB_FALSE);
686 
687  for (c = qb_ipcs_connection_first_get(ipcs_mapper[service_id].inst);
688  c;
689  prev = c, c = qb_ipcs_connection_next_get(ipcs_mapper[service_id].inst, prev), qb_ipcs_connection_unref(prev)) {
690 
691  cnx = qb_ipcs_context_get(c);
692  if (cnx == NULL) continue;
693  if (c != conn_ptr) continue;
694 
695  qb_ipcs_connection_stats_get(c, &ipcs_stats->conn, QB_FALSE);
696  if (ipcs_stats->conn.client_pid != pid) {
697  continue;
698  }
699  found = 1;
700  memcpy(&ipcs_stats->cnx, cnx, sizeof(struct cs_ipcs_conn_context));
701  }
702  if (!found) {
703  return CS_ERR_NOT_EXIST;
704  }
705 
706  return CS_OK;
707 }
708 
710 {
711  struct cs_ipcs_conn_context *cnx;
712  struct ipcs_conn_stats ipcs_stats;
713  qb_ipcs_connection_t *c, *prev;
714  int service_id;
715 
716  /* Global stats are easy */
717  memset(&global_stats, 0, sizeof(global_stats));
718 
719  for (service_id = 0; service_id < SERVICES_COUNT_MAX; service_id++) {
720  if (!ipcs_mapper[service_id].inst) {
721  continue;
722  }
723 
724  for (c = qb_ipcs_connection_first_get(ipcs_mapper[service_id].inst);
725  c;
726  prev = c, c = qb_ipcs_connection_next_get(ipcs_mapper[service_id].inst, prev), qb_ipcs_connection_unref(prev)) {
727  /* Get stats with 'clear_after_read' set */
728  qb_ipcs_connection_stats_get(c, &ipcs_stats.conn, QB_TRUE);
729 
730  /* Our own stats */
731  cnx = qb_ipcs_context_get(c);
732  if (cnx == NULL) continue;
733  cnx->invalid_request = 0;
734  cnx->overload = 0;
735  cnx->sent = 0;
736 
737  }
738  }
739 }
740 
741 static enum qb_ipc_type cs_get_ipc_type (void)
742 {
743  char *str;
744  int found = 0;
745  enum qb_ipc_type ret = QB_IPC_NATIVE;
746 
747  if (icmap_get_string("system.qb_ipc_type", &str) != CS_OK) {
748  log_printf(LOGSYS_LEVEL_DEBUG, "No configured system.qb_ipc_type. Using native ipc");
749  return QB_IPC_NATIVE;
750  }
751 
752  if (strcmp(str, "native") == 0) {
753  ret = QB_IPC_NATIVE;
754  found = 1;
755  }
756 
757  if (strcmp(str, "shm") == 0) {
758  ret = QB_IPC_SHM;
759  found = 1;
760  }
761 
762  if (strcmp(str, "socket") == 0) {
763  ret = QB_IPC_SOCKET;
764  found = 1;
765  }
766 
767  if (found) {
768  log_printf(LOGSYS_LEVEL_DEBUG, "Using %s ipc", str);
769  } else {
770  log_printf(LOGSYS_LEVEL_DEBUG, "Unknown ipc type %s", str);
771  }
772 
773  free(str);
774 
775  return ret;
776 }
777 
778 const char *cs_ipcs_service_init(struct corosync_service_engine *service)
779 {
780  const char *serv_short_name;
781 
782  serv_short_name = cs_ipcs_serv_short_name(service->id);
783 
784  if (service->lib_engine_count == 0) {
786  "NOT Initializing IPC on %s [%d]",
787  serv_short_name,
788  service->id);
789  return NULL;
790  }
791 
792  if (strlen(serv_short_name) >= CS_IPCS_MAPPER_SERV_NAME) {
793  log_printf (LOGSYS_LEVEL_ERROR, "service name %s is too long", serv_short_name);
794  return "qb_ipcs_run error";
795  }
796 
797  ipcs_mapper[service->id].id = service->id;
798  strcpy(ipcs_mapper[service->id].name, serv_short_name);
800  "Initializing IPC on %s [%d]",
801  ipcs_mapper[service->id].name,
802  ipcs_mapper[service->id].id);
803  ipcs_mapper[service->id].inst = qb_ipcs_create(ipcs_mapper[service->id].name,
804  ipcs_mapper[service->id].id,
805  cs_get_ipc_type(),
806  &corosync_service_funcs);
807  assert(ipcs_mapper[service->id].inst);
808  qb_ipcs_poll_handlers_set(ipcs_mapper[service->id].inst,
809  &corosync_poll_funcs);
810  if (qb_ipcs_run(ipcs_mapper[service->id].inst) != 0) {
811  log_printf (LOGSYS_LEVEL_ERROR, "Can't initialize IPC");
812  return "qb_ipcs_run error";
813  }
814 
815  return NULL;
816 }
817 
818 void cs_ipcs_init(void)
819 {
820  api = apidef_get ();
821 
822  qb_loop_poll_low_fds_event_set(cs_poll_handle_get(), cs_ipcs_low_fds_event);
823 
824  api->quorum_register_callback (cs_ipcs_fc_quorum_changed, NULL);
825  totempg_queue_level_register_callback (cs_ipcs_totem_queue_level_changed);
826 
827  global_stats.active = 0;
828  global_stats.closed = 0;
829 }
corosync_api_v1
The corosync_api_v1 struct.
Definition: coroapi.h:225
cs_ipcs_conn_context::overload
uint64_t overload
Definition: ipcs_stats.h:40
apidef.h
corosync_sending_allowed_release
void corosync_sending_allowed_release(void *sending_allowed_private_data)
Definition: main.c:776
outq_item::msg
void * msg
Definition: ipc_glue.c:84
CS_LIB_ALLOW_INQUORATE
Definition: coroapi.h:164
cs_ipcs_conn_context::proc_name
char proc_name[32]
Definition: ipcs_stats.h:42
cs_ipcs_dispatch_iov_send
int cs_ipcs_dispatch_iov_send(void *conn, const struct iovec *iov, unsigned int iov_len)
Definition: ipc_glue.c:483
corosync_service_engine::lib_exit_fn
int(* lib_exit_fn)(void *conn)
Definition: coroapi.h:503
outq_item::list
struct qb_list_head list
Definition: ipc_glue.c:86
log_printf
#define log_printf(level, format, args...)
Definition: logsys.h:323
corosync_service_engine::id
unsigned short id
Definition: coroapi.h:492
stats_ipcs_del_connection
void stats_ipcs_del_connection(int service_id, uint32_t pid, void *ptr)
Definition: stats.c:659
totempg.h
stats_ipcs_add_connection
void stats_ipcs_add_connection(int service_id, uint32_t pid, void *ptr)
Definition: stats.c:649
service.h
ipcs_stats.h
CPG_SERVICE
Definition: corodefs.h:46
sync.h
outq_item::mlen
size_t mlen
Definition: ipc_glue.c:85
CS_ERR_TRY_AGAIN
Definition: corotypes.h:100
LOGSYS_LEVEL_ERROR
#define LOGSYS_LEVEL_ERROR
Definition: logsys.h:72
TOTEM_Q_LEVEL_LOW
Definition: totempg.h:178
cs_ipcs_q_level_get
int32_t cs_ipcs_q_level_get(void)
Definition: ipc_glue.c:601
LOGSYS_LEVEL_DEBUG
#define LOGSYS_LEVEL_DEBUG
Definition: logsys.h:76
ipcs_global_stats::active
uint64_t active
Definition: ipcs_stats.h:48
CS_ERR_NOT_EXIST
Definition: corotypes.h:106
corosync_service
struct corosync_service_engine * corosync_service[SERVICES_COUNT_MAX]
Definition: service.c:110
cs_ipcs_conn_context::sent
uint32_t sent
Definition: ipcs_stats.h:41
cs_ipcs_conn_context::queuing
int32_t queuing
Definition: ipcs_stats.h:37
ipcs_conn_stats::conn
struct qb_ipcs_connection_stats conn
Definition: ipcs_stats.h:55
cs_ipcs_dispatch_send
int cs_ipcs_dispatch_send(void *conn, const void *msg, size_t mlen)
Definition: ipc_glue.c:474
quorate
uint32_t quorate
Definition: sam.c:134
cs_ipcs_service_init
const char * cs_ipcs_service_init(struct corosync_service_engine *service)
Definition: ipc_glue.c:778
cs_ipcs_response_iov_send
int cs_ipcs_response_iov_send(void *conn, const struct iovec *iov, unsigned int iov_len)
Definition: ipc_glue.c:363
cs_ipcs_conn_context
Definition: ipcs_stats.h:35
swab.h
cs_poll_handle_get
qb_loop_t * cs_poll_handle_get(void)
Definition: main.c:170
ipcs_global_stats
Definition: ipcs_stats.h:46
cs_ipcs_conn_context::data
char data[1]
Definition: ipcs_stats.h:43
cs_ipcs_mapper::inst
qb_ipcs_service_t * inst
Definition: ipc_glue.c:79
corosync_service_engine
The corosync_service_engine struct.
Definition: coroapi.h:490
icmap.h
stats.h
CS_OK
Definition: corotypes.h:95
LOGSYS_LEVEL_INFO
#define LOGSYS_LEVEL_INFO
Definition: logsys.h:75
CS_IPCS_MAPPER_SERV_NAME
#define CS_IPCS_MAPPER_SERV_NAME
Definition: ipc_glue.c:75
corosync_service_engine::lib_engine
struct corosync_lib_handler * lib_engine
Definition: coroapi.h:504
cs_ipcs_private_data_get
void * cs_ipcs_private_data_get(void *conn)
Definition: ipc_glue.c:312
corosync_service_engine::lib_engine_count
int lib_engine_count
Definition: coroapi.h:505
ICMAP_KEYNAME_MAXLEN
#define ICMAP_KEYNAME_MAXLEN
Maximum length of key in icmap.
Definition: icmap.h:48
timer.h
cs_error_t
cs_error_t
The cs_error_t enum.
Definition: corotypes.h:94
TOTEM_Q_LEVEL_CRITICAL
Definition: totempg.h:181
ipcs_conn_stats::cnx
struct cs_ipcs_conn_context cnx
Definition: ipcs_stats.h:56
cs_ipcs_response_send
int cs_ipcs_response_send(void *conn, const void *msg, size_t mlen)
Definition: ipc_glue.c:374
ipcs_conn_stats
Definition: ipcs_stats.h:52
cs_ipcs_conn_context::outq_head
struct qb_list_head outq_head
Definition: ipcs_stats.h:36
cs_ipc_allow_connections
void cs_ipc_allow_connections(int32_t allow)
Definition: ipc_glue.c:159
CFG_SERVICE
Definition: corodefs.h:45
corosync_api_v1::quorum_register_callback
int(* quorum_register_callback)(quorum_callback_fn_t callback_fn, void *context)
Definition: coroapi.h:386
LOGSYS_LEVEL_NOTICE
#define LOGSYS_LEVEL_NOTICE
Definition: logsys.h:74
corosync_service_engine::private_data_size
size_t private_data_size
Definition: coroapi.h:496
PLOAD_SERVICE
Definition: corodefs.h:48
cs_ipcs_conn_context::queued
uint32_t queued
Definition: ipcs_stats.h:38
corosync_sending_allowed
int corosync_sending_allowed(unsigned int service, unsigned int id, const void *msg, void *sending_allowed_private_data)
Definition: main.c:728
MON_SERVICE
Definition: corodefs.h:50
ipcs_conn_stats::srv
struct qb_ipcs_stats srv
Definition: ipcs_stats.h:54
cs_ipcs_clear_stats
void cs_ipcs_clear_stats()
Definition: ipc_glue.c:709
cs_ipcs_conn_context::invalid_request
uint64_t invalid_request
Definition: ipcs_stats.h:39
cs_ipc_refcnt_inc
void cs_ipc_refcnt_inc(void *conn)
Definition: ipc_glue.c:302
cs_ipcs_init
void cs_ipcs_init(void)
Definition: ipc_glue.c:818
VOTEQUORUM_SERVICE
Definition: corodefs.h:49
icmap_get_string
cs_error_t icmap_get_string(const char *key_name, char **str)
Shortcut for icmap_get for string type.
Definition: icmap.c:880
SERVICES_COUNT_MAX
#define SERVICES_COUNT_MAX
Definition: coroapi.h:462
main.h
util.h
outq_item
Definition: ipc_glue.c:83
cs_ipcs_sync_state_changed
void cs_ipcs_sync_state_changed(int32_t sync_in_process)
Definition: ipc_glue.c:664
corosync_lib_handler::lib_handler_fn
void(* lib_handler_fn)(void *conn, const void *msg)
Definition: coroapi.h:468
cs_ipcs_service_destroy
int32_t cs_ipcs_service_destroy(int32_t service_id)
Definition: ipc_glue.c:164
LOGSYS_DECLARE_SUBSYS
LOGSYS_DECLARE_SUBSYS("MAIN")
cs_ipc_refcnt_dec
void cs_ipc_refcnt_dec(void *conn)
Definition: ipc_glue.c:307
config.h
cs_ipcs_get_global_stats
void cs_ipcs_get_global_stats(struct ipcs_global_stats *ipcs_stats)
Definition: ipc_glue.c:670
CS_ERR_INVALID_PARAM
Definition: corotypes.h:101
logsys.h
TOTEM_Q_LEVEL_HIGH
Definition: totempg.h:180
totem_q_level
totem_q_level
Definition: totempg.h:177
ipcs_global_stats::closed
uint64_t closed
Definition: ipcs_stats.h:49
corotypes.h
LOGSYS_LEVEL_WARNING
#define LOGSYS_LEVEL_WARNING
Definition: logsys.h:73
QUORUM_SERVICE
Definition: corodefs.h:47
totempg_queue_level_register_callback
void totempg_queue_level_register_callback(totem_queue_level_changed_fn)
Definition: totempg.c:1510
apidef_get
struct corosync_api_v1 * apidef_get(void)
Definition: apidef.c:146
cs_ipcs_get_conn_stats
cs_error_t cs_ipcs_get_conn_stats(int service_id, uint32_t pid, void *conn_ptr, struct ipcs_conn_stats *ipcs_stats)
Definition: ipc_glue.c:675
corosync_recheck_the_q_level
void corosync_recheck_the_q_level(void *data)
Definition: main.c:714
cs_ipcs_mapper::id
int32_t id
Definition: ipc_glue.c:78
CMAP_SERVICE
Definition: corodefs.h:44
cs_ipcs_mapper::name
char name[CS_IPCS_MAPPER_SERV_NAME]
Definition: ipc_glue.c:80
WD_SERVICE
Definition: corodefs.h:51
icmap_get_uint8
cs_error_t icmap_get_uint8(const char *key_name, uint8_t *u8)
Definition: icmap.c:826
cs_ipcs_mapper
Definition: ipc_glue.c:77
TOTEM_Q_LEVEL_GOOD
Definition: totempg.h:179
corodefs.h