src/status.c

   1 /* vim:tw=78:ts=8:sw=4:set ft=c:  */
   2 /*
   3     Copyright (C) 2006-2009 Ben Kibbey <bjk@luxsci.net>
   4
   5     This program is free software; you can redistribute it and/or modify
   6     it under the terms of the GNU General Public License as published by
   7     the Free Software Foundation; either version 2 of the License, or
   8     (at your option) any later version.
   9
  10     This program is distributed in the hope that it will be useful,
  11     but WITHOUT ANY WARRANTY; without even the implied warranty of
  12     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13     GNU General Public License for more details.
  14
  15     You should have received a copy of the GNU General Public License
  16     along with this program; if not, write to the Free Software
  17     Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02110-1301  USA
  18 */
  19 #include <glib.h>
  20 #include "common.h"
  21 #include "lock.h"
  22 #include "misc.h"
  23 #include "status.h"
  24 #include "pwmd_error.h"
  25
  26 struct status_thread_s {
  27     assuan_context_t ctx;
  28     gchar *status;
  29     const gchar *line;
  30     pthread_t tid;
  31     gint fd[2];
  32 };
  33
  34 static void cleanup(void *arg)
  35 {
  36     struct status_thread_s *s = arg;
  37
  38     close(s->fd[0]);
  39     close(s->fd[1]);
  40     g_free(s);
  41 }
  42
  43 static void *send_status_thread(void *arg)
  44 {
  45     struct status_thread_s *s = arg;
  46     gpg_error_t rc;
  47
  48     pthread_cleanup_push(cleanup, s);
  49     pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);
  50     rc = assuan_write_status(s->ctx, s->status, s->line);
  51     pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL);
  52     write(s->fd[1], &rc, sizeof(gpg_error_t));
  53     pthread_cleanup_pop(0);
  54     return NULL;
  55 }
  56
  57 gpg_error_t send_status(assuan_context_t ctx, status_msg_t which,
  58         const gchar *fmt, ...)
  59 {
  60     const gchar *line = NULL;
  61     struct client_s *client = assuan_get_pointer(ctx);
  62     gchar buf[ASSUAN_LINELENGTH];
  63     gchar *status = NULL;
  64     va_list ap;
  65     gint n;
  66     gpg_error_t rc;
  67     struct status_thread_s *s;
  68     struct timeval tv = {0, 0};
  69     gint to = get_key_file_integer("global", "keepalive");
  70     fd_set rfds;
  71     pthread_attr_t attr;
  72
  73     if (fmt) {
  74         va_start(ap, fmt);
  75         g_vsnprintf(buf, sizeof(buf), fmt, ap);
  76         va_end(ap);
  77         line = buf;
  78     }
  79
  80     switch (which) {
  81         case STATUS_CACHE:
  82             CACHE_LOCK(client->ctx);
  83             line = print_fmt(buf, sizeof(buf), "%i", cache_file_count());
  84             CACHE_UNLOCK;
  85             status = "CACHE";
  86             break;
  87         case STATUS_CLIENTS:
  88             MUTEX_LOCK(&cn_mutex);
  89             line = print_fmt(buf, sizeof(buf), "%i", g_slist_length(cn_thread_list));
  90             MUTEX_UNLOCK(&cn_mutex);
  91             status = "CLIENTS";
  92             break;
  93         case STATUS_CONFIG:
  94             status = "CONFIG";
  95             break;
  96         case STATUS_KEEPALIVE:
  97             status = "KEEPALIVE";
  98             break;
  99         case STATUS_LOCKED:
 100             status = "LOCKED";
 101             line = N_("Waiting for lock");
 102             break;
 103         case STATUS_ENCRYPT:
 104             status = "ENCRYPT";
 105             break;
 106         case STATUS_DECRYPT:
 107             status = "DECRYPT";
 108             break;
 109         case STATUS_DECOMPRESS:
 110             status = "DECOMPRESS";
 111             break;
 112         case STATUS_COMPRESS:
 113             status = "COMPRESS";
 114             break;
 115     }
 116
 117     if (!ctx) {
 118         log_write("%s %s", status, line);
 119         return 0;
 120     }
 121
 122     s = g_malloc0(sizeof(struct status_thread_s));
 123
 124     if (!s) {
 125         log_write("%s(%i): %s", __FILE__, __LINE__, strerror(ENOMEM));
 126         return gpg_error_from_errno(ENOMEM);
 127     }
 128
 129     s->ctx = ctx;
 130     s->status = status;
 131     s->line = line;
 132
 133     /* Since we use the keepalive from the configuration, it may be 0. If so,
 134      * status messages would fail and abort the connection. So use a default
 135      * that doesn't affect the configured keepalive value. */
 136     tv.tv_sec = to <= 0 ? DEFAULT_KEEPALIVE_TO : to;
 137
 138     if (pipe(s->fd) == -1) {
 139         n = errno;
 140         g_free(s);
 141         return gpg_error_from_errno(n);
 142     }
 143
 144     pthread_cleanup_push(cleanup, s);
 145     pthread_attr_init(&attr);
 146     pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
 147     pthread_create(&s->tid, &attr, send_status_thread, s);
 148     pthread_attr_destroy(&attr);
 149     FD_ZERO(&rfds);
 150     FD_SET(s->fd[0], &rfds);
 151     n = s->fd[0] > s->fd[1] ? s->fd[0] : s->fd[1];
 152     n = select(n+1, &rfds, NULL, NULL, &tv);
 153
 154     if (n <= 0) {
 155         n = errno;
 156         pthread_cancel(s->tid);
 157
 158         if (n) {
 159             log_write("%s(%i): %s", __FILE__, __LINE__, strerror(n));
 160             rc = gpg_error_from_errno(n);
 161         }
 162         else
 163             rc = GPG_ERR_ASS_WRITE_ERROR;
 164
 165         return rc;
 166     }
 167
 168     if (FD_ISSET(s->fd[0], &rfds)) {
 169         size_t len = read(s->fd[0], &rc, sizeof(gpg_error_t));
 170
 171         if (len != sizeof(gpg_error_t)) {
 172             log_write("%s(%i): %s", __FILE__, __LINE__, strerror(errno));
 173             rc = GPG_ERR_ASS_WRITE_ERROR;
 174         }
 175     }
 176     else
 177         rc = GPG_ERR_ASS_WRITE_ERROR;
 178
 179     pthread_cleanup_pop(1);
 180     return rc;
 181 }
 182
 183 /* This is needed for FreeBSD systems (and maybe others). I'm guessing here,
 184  * but I think it lets the scheduler let other threads do their work. If we
 185  * were to do the write() in send_status_all() then write() would return
 186  * before the signal was actually sent to client_msg_thread(). This is only
 187  * called from send_status_all().
 188  */
 189 static void *write_status_msg(void *arg)
 190 {
 191     struct client_thread_s *cn  =arg;
 192     size_t len;
 193
 194     pthread_cond_signal(&cn->msg_cond);
 195     len = write(cn->msg_fd[1], &cn->msg, sizeof(status_msg_t));
 196
 197     if (len != sizeof(status_msg_t))
 198         log_write("%s(%i): %s", __FILE__, __LINE__, strerror(errno));
 199
 200     return NULL;
 201 }
 202
 203 void send_status_all(status_msg_t which)
 204 {
 205     guint i, t;
 206
 207     MUTEX_LOCK(&cn_mutex);
 208
 209     for (t = g_slist_length(cn_thread_list), i = 0; i < t; i++) {
 210         struct client_thread_s *cn = g_slist_nth_data(cn_thread_list, i);
 211         pthread_t tid;
 212
 213         if (cn->quit)
 214             continue;
 215
 216         cn->msg = which;
 217         pthread_create(&tid, NULL, write_status_msg, cn);
 218         pthread_join(tid, NULL);
 219     }
 220
 221     MUTEX_UNLOCK(&cn_mutex);
 222 }
 223
 224 /* The msg_sender_mutex should have been locked before calling this thread
 225  * entry point. */
 226 void *client_msg_sender_thread(void *arg)
 227 {
 228     struct client_thread_s *thd = arg;
 229
 230     for (;;) {
 231         struct status_msg_s *msg;
 232         gpg_error_t rc;
 233
 234         pthread_cond_wait(&thd->msg_sender_cond, &thd->msg_sender_mutex);
 235         pthread_testcancel();
 236
 237         /* The messages may have been stacked while waiting for send_status()
 238          * to return. Send what's in the queue. */
 239         for (;;) {
 240             msg = g_slist_nth_data(thd->msg_queue, 0);
 241
 242             if (!msg)
 243                 break;
 244
 245             /* Unlock to prevent blocking in client_msg_thread(). */
 246             MUTEX_UNLOCK(&thd->msg_sender_mutex);
 247             rc = send_status(thd->cl->ctx, msg->msg, NULL);
 248             MUTEX_LOCK(&thd->msg_sender_mutex);
 249             thd->msg_queue = g_slist_remove(thd->msg_queue, msg);
 250             g_free(msg);
 251             pthread_testcancel();
 252
 253             if (rc) {
 254                 log_write(N_("msg for %i failed: %s"), thd->fd,
 255                         pwmd_strerror(rc));
 256                 pthread_cancel(thd->tid);
 257                 break;
 258             }
 259         }
 260     }
 261
 262     return NULL;
 263 }
 264
 265 /*
 266  * This function waits for a signal from send_status_all() then appends a
 267  * message read from a pipe to the clients message queue. The actual sending
 268  * of the message is done in client_msg_sender_thread() which waits for a
 269  * signal from this function. This prevents blocking in assuan_send_status()
 270  * when sending to remote clients. This also avoids duplicate status messages.
 271  * If an existing status message in the message queue has the same type as the
 272  * current message the the current one will be skipped. This avoids flooding
 273  * the client with old status messages.
 274  */
 275 void *client_msg_thread(void *arg)
 276 {
 277     struct client_thread_s *thd = arg;
 278     pthread_mutex_t mu;
 279
 280     pthread_mutex_init(&mu, NULL);
 281     pthread_mutex_lock(&mu);
 282
 283     for (;;) {
 284         fd_set rfds;
 285         int n;
 286         status_msg_t m;
 287         size_t len;
 288         struct status_msg_s *msg;
 289
 290         pthread_cleanup_push(pthread_mutex_destroy, &mu);
 291         pthread_cleanup_push(pthread_mutex_unlock, &mu);
 292         pthread_cond_wait(&thd->msg_cond, &mu);
 293         pthread_testcancel();
 294         FD_ZERO(&rfds);
 295         FD_SET(thd->msg_fd[0], &rfds);
 296         n = select(thd->msg_fd[0]+1, &rfds, NULL, NULL, NULL);
 297         pthread_testcancel();
 298
 299         if (n <= 0 || !FD_ISSET(thd->msg_fd[0], &rfds))
 300             continue;
 301
 302         len = read(thd->msg_fd[0], &m, sizeof(status_msg_t));
 303         pthread_testcancel();
 304
 305         if (len != sizeof(status_msg_t)) {
 306             log_write("%s(%i): %s", __FILE__, __LINE__, strerror(errno));
 307             continue;
 308         }
 309
 310         MUTEX_LOCK(&thd->msg_sender_mutex);
 311
 312         for (n = 0; n < g_slist_length(thd->msg_queue); n++) {
 313             msg = g_slist_nth_data(thd->msg_queue, n);
 314
 315             if (msg->msg == m)
 316                 goto done;
 317         }
 318
 319         msg = g_malloc(sizeof(struct status_msg_s));
 320
 321         if (!msg) {
 322             log_write("%s(%i): %s", __FILE__, __LINE__, strerror(ENOMEM));
 323             MUTEX_UNLOCK(&thd->msg_sender_mutex);
 324             continue;
 325         }
 326
 327         msg->msg = m;
 328         thd->msg_queue = g_slist_append(thd->msg_queue, msg);
 329 done:
 330         MUTEX_UNLOCK(&thd->msg_sender_mutex);
 331         pthread_cond_signal(&thd->msg_sender_cond);
 332         pthread_cleanup_pop(0);
 333         pthread_cleanup_pop(0);
 334     }
 335
 336     return NULL;
 337 }