svc_dev.c

   1 /*
   2  * Copyright (C) 2012-2020 all contributors <cmogstored-public@yhbt.net>
   3  * License: GPL-3.0+ <https://www.gnu.org/licenses/gpl-3.0.txt>
   4  */
   5 #include "cmogstored.h"
   6 #include "compat_memstream.h"
   7
   8 /*
   9  * maps multiple "devXXX" directories to the device.
  10  * This is uncommon in real world deployments (multiple mogdevs sharing
  11  * the same system device), but happens frequently in testing
  12  */
  13 struct mog_devlist {
  14         dev_t st_dev;
  15         Hash_table *by_mogdevid;
  16 };
  17
  18 static size_t devlist_hash(const void *x, size_t tablesize)
  19 {
  20         const struct mog_devlist *devlist = x;
  21
  22         return devlist->st_dev % tablesize;
  23 }
  24
  25 static bool devlist_cmp(const void *a, const void *b)
  26 {
  27         const struct mog_devlist *devlist_a = a;
  28         const struct mog_devlist *devlist_b = b;
  29
  30         return devlist_a->st_dev == devlist_b->st_dev;
  31 }
  32
  33 static void devlist_free(void *x)
  34 {
  35         struct mog_devlist *devlist = x;
  36
  37         hash_free(devlist->by_mogdevid);
  38         free(devlist);
  39 }
  40
  41 static struct mog_devlist * mog_devlist_new(dev_t st_dev)
  42 {
  43         struct mog_devlist *devlist = xmalloc(sizeof(struct mog_devlist));
  44
  45         devlist->st_dev = st_dev;
  46         devlist->by_mogdevid = hash_initialize(7, NULL,
  47                                                mog_dev_hash, mog_dev_cmp,
  48
  49                                                /*
  50                                                 * elements are freed when
  51                                                 * svc->by_mog_devid is freed
  52                                                 */
  53                                                NULL);
  54
  55         mog_oom_if_null(devlist->by_mogdevid);
  56
  57         return devlist;
  58 }
  59
  60 /* ensures svc has a devlist, this must be called with devstats_lock held */
  61 static struct mog_devlist * svc_devlist(struct mog_svc *svc, dev_t st_dev)
  62 {
  63         struct mog_devlist *devlist;
  64         struct mog_devlist finder;
  65
  66         assert(svc->by_st_dev && "by_st_dev unintialized in svc");
  67
  68         finder.st_dev = st_dev;
  69         devlist = hash_lookup(svc->by_st_dev, &finder);
  70
  71         if (devlist == NULL) {
  72                 devlist = mog_devlist_new(st_dev);
  73                 switch (hash_insert_if_absent(svc->by_st_dev, devlist, NULL)) {
  74                 case 0:
  75                         assert(0 && "race condition, devlist should insert "
  76                                "without error");
  77                         abort();
  78                 case 1: break; /* OK, inserted */
  79                 default: mog_oom(); /* -1 */
  80                 }
  81         }
  82         return devlist;
  83 }
  84
  85 static void svc_init_dev_hash(struct mog_svc *svc)
  86 {
  87         if (svc->by_st_dev) {
  88                 hash_clear(svc->by_st_dev);
  89                 return;
  90         }
  91
  92         svc->by_st_dev = hash_initialize(7, NULL, devlist_hash,
  93                                          devlist_cmp, devlist_free);
  94         mog_oom_if_null(svc->by_st_dev);
  95 }
  96
  97 static int svc_scandev(struct mog_svc *svc, unsigned *nr, mog_scandev_cb cb)
  98 {
  99         struct dirent *ent;
 100         int rc = 0;
 101
 102         CHECK(int, 0, pthread_mutex_lock(&svc->devstats_lock));
 103         svc_init_dev_hash(svc);
 104         rewinddir(svc->dir);
 105         while ((ent = readdir(svc->dir))) {
 106                 unsigned long mog_devid;
 107                 char *end;
 108                 size_t len = strlen(ent->d_name);
 109                 struct mog_dev *dev;
 110                 struct mog_devlist *devlist;
 111                 Hash_table *devhash;
 112
 113                 if (len <= 3) continue;
 114                 if (memcmp("dev", ent->d_name, 3) != 0) continue;
 115
 116                 mog_devid = strtoul(ent->d_name + 3, &end, 10);
 117                 if (*end != 0) continue;
 118                 if (mog_devid > MOG_DEVID_MAX) continue;
 119
 120                 dev = mog_dev_for(svc, (uint32_t)mog_devid, true);
 121                 if (!dev) continue;
 122
 123                 devlist = svc_devlist(svc, dev->st_dev);
 124                 devhash = devlist->by_mogdevid;
 125
 126                 if (cb)
 127                         rc |= cb(dev, svc); /* mog_dev_mkusage */
 128
 129                 switch (hash_insert_if_absent(devhash, dev, NULL)) {
 130                 case 0:
 131                         /* do not free dev, it is in svc->by_mog_devid */
 132                         syslog(LOG_ERR,
 133                                "%s/%s seen twice in readdir (BUG?)",
 134                                svc->docroot, ent->d_name);
 135                         break;
 136                 case 1:
 137                         (*nr)++;
 138                         break;
 139                 default: mog_oom(); /* -1 */
 140                 }
 141         }
 142         CHECK(int, 0, pthread_mutex_unlock(&svc->devstats_lock));
 143
 144         return rc;
 145 }
 146
 147 static bool write_dev_stats(void *entry, void *filep)
 148 {
 149         struct mog_dev *dev = entry;
 150         FILE **fp = filep;
 151         char util[MOG_IOUTIL_LEN];
 152
 153         mog_iou_read(dev->st_dev, util);
 154
 155         if (fprintf(*fp, "%u\t%s\n", dev->devid, util) > 0)
 156                 return true;
 157
 158         /* stop iteration in case we get EIO/ENOSPC on systems w/o memstream */
 159         my_memstream_errclose(*fp);
 160         *fp = NULL;
 161         return false;
 162 }
 163
 164 static bool write_devlist_stats(void *entry, void *filep)
 165 {
 166         struct mog_devlist *devlist = entry;
 167         FILE **fp ;
 168
 169         hash_do_for_each(devlist->by_mogdevid, write_dev_stats, filep);
 170         fp = filep;
 171
 172         /* *filep becomes NULL on errors */
 173         return !!*fp;
 174 }
 175
 176 /* updates per-svc device stats from the global mount list */
 177 static ssize_t devstats_stringify(struct mog_svc *svc, char **dst)
 178 {
 179         FILE *fp;
 180         size_t bytes;
 181
 182         assert(svc->by_st_dev && "need to scan devices first");
 183
 184         /* open_memstream() may fail on EIO/EMFILE/ENFILE on fake memstream */
 185         fp = open_memstream(dst, &bytes);
 186         if (!fp)
 187                 return -1;
 188
 189         /*
 190          * write_devlist_stats->write_dev_stats may fclose and NULL fp
 191          * to indicate error:
 192          */
 193         hash_do_for_each(svc->by_st_dev, write_devlist_stats, &fp);
 194         if (!fp)
 195                 return -1;
 196
 197         if (fprintf(fp, ".\n") == 2) {
 198                 CHECK(int, 0, my_memstream_close(fp, dst, &bytes));
 199                 return bytes;
 200         }
 201
 202         my_memstream_errclose(fp);
 203         return -1;
 204 }
 205
 206 void mog_svc_devstats_subscribe(struct mog_mgmt *mgmt)
 207 {
 208         struct mog_svc *svc = mgmt->svc;
 209
 210         CHECK(int, 0, pthread_mutex_lock(&svc->devstats_lock));
 211         LIST_INSERT_HEAD(&svc->devstats_subscribers, mgmt, subscribed);
 212         CHECK(int, 0, pthread_mutex_unlock(&svc->devstats_lock));
 213 }
 214
 215 /* called while iterating through all mog_svc objects */
 216 bool mog_svc_devstats_broadcast(void *ent, void *ignored)
 217 {
 218         struct mog_svc *svc = ent;
 219         struct mog_mgmt *mgmt, *tmp;
 220         struct iovec iov;
 221         char *buf = NULL;
 222         ssize_t len;
 223         struct mog_fd *mfd;
 224
 225         CHECK(int, 0, pthread_mutex_lock(&svc->devstats_lock));
 226
 227         len = devstats_stringify(svc, &buf);
 228         if (len < 0)
 229                 goto out;
 230
 231         LIST_FOREACH_SAFE(mgmt, &svc->devstats_subscribers, subscribed, tmp) {
 232                 assert(mgmt->wbuf == NULL && "wbuf not null");
 233                 iov.iov_base = buf;
 234                 iov.iov_len = (size_t)len;
 235                 mog_mgmt_writev(mgmt, &iov, 1);
 236
 237                 if (mgmt->wbuf == NULL) continue; /* success */
 238
 239                 LIST_REMOVE(mgmt, subscribed);
 240                 mfd = mog_fd_of(mgmt);
 241                 if (mgmt->wbuf == MOG_WR_ERROR) {
 242                         assert(mgmt->rbuf == NULL && "would leak rbuf");
 243                         mog_fd_put(mfd);
 244                 } else { /* blocked on write */
 245                         mog_idleq_push(mgmt->svc->queue, mfd, MOG_QEV_WR);
 246                 }
 247         }
 248 out:
 249         CHECK(int, 0, pthread_mutex_unlock(&svc->devstats_lock));
 250
 251         free(buf);
 252
 253         return true;
 254 }
 255
 256 static bool devstats_shutdown_i(void *svcptr, void *ignored)
 257 {
 258         struct mog_svc *svc = svcptr;
 259         struct mog_mgmt *mgmt, *tmp;
 260         struct mog_fd *mfd;
 261
 262         CHECK(int, 0, pthread_mutex_lock(&svc->devstats_lock));
 263         LIST_FOREACH_SAFE(mgmt, &svc->devstats_subscribers, subscribed, tmp) {
 264                 assert(mgmt->wbuf == NULL && "wbuf not null");
 265                 assert(mgmt->rbuf == NULL && "would leak rbuf");
 266                 LIST_REMOVE(mgmt, subscribed);
 267                 mfd = mog_fd_of(mgmt);
 268                 mog_fd_put(mfd);
 269         }
 270         CHECK(int, 0, pthread_mutex_unlock(&svc->devstats_lock));
 271
 272         return true;
 273 }
 274
 275 void mog_svc_dev_shutdown(void)
 276 {
 277         mog_svc_each(devstats_shutdown_i, NULL);
 278 }
 279
 280 static bool svc_mkusage_each(void *svcptr, void *ignored)
 281 {
 282         struct mog_svc *svc = svcptr;
 283         unsigned ndev = 0;
 284
 285         svc_scandev(svc, &ndev, mog_dev_mkusage);
 286
 287         if (svc->queue && (svc->nmogdev != ndev))
 288                 mog_svc_thrpool_rescale(svc, ndev);
 289         svc->nmogdev = ndev;
 290
 291         return true;
 292 }
 293
 294 void mog_mkusage_all(void)
 295 {
 296         mog_svc_each(svc_mkusage_each, NULL);
 297 }
 298
 299 /* we should never set ioq_max == 0 */
 300 static void svc_rescale_warn_fix_capa(struct mog_svc *svc, unsigned ndev_new)
 301 {
 302         if (svc->thr_per_dev != 0)
 303                 return;
 304
 305         syslog(LOG_WARNING,
 306                "serving %s with fewer aio_threads(%u) than devices(%u)",
 307                svc->docroot, svc->user_set_aio_threads, ndev_new);
 308         syslog(LOG_WARNING,
 309                "set \"server aio_threads = %u\" or higher via sidechannel",
 310                ndev_new);
 311
 312         svc->thr_per_dev = 1;
 313 }
 314
 315 static void mog_svc_dev_rescale_all(struct mog_svc *svc)
 316 {
 317         /* iterate through each device of this svc */
 318         CHECK(int, 0, pthread_mutex_lock(&svc->by_mog_devid_lock));
 319         hash_do_for_each(svc->by_mog_devid, mog_dev_user_rescale_i, svc);
 320         CHECK(int, 0, pthread_mutex_unlock(&svc->by_mog_devid_lock));
 321 }
 322
 323 void mog_svc_dev_requeue_prepare(struct mog_svc *svc)
 324 {
 325         /* iterate through each device of this svc */
 326         CHECK(int, 0, pthread_mutex_lock(&svc->by_mog_devid_lock));
 327         hash_do_for_each(svc->by_mog_devid, mog_dev_requeue_prepare, svc);
 328         CHECK(int, 0, pthread_mutex_unlock(&svc->by_mog_devid_lock));
 329 }
 330
 331 /* rescaling only happens in the main thread */
 332 void mog_svc_dev_user_rescale(struct mog_svc *svc, size_t ndev_new)
 333 {
 334         assert(svc->user_set_aio_threads &&
 335                "user did not set aio_threads via sidechannel");
 336
 337         svc->thr_per_dev = svc->user_set_aio_threads / ndev_new;
 338
 339         svc_rescale_warn_fix_capa(svc, ndev_new);
 340         mog_svc_dev_rescale_all(svc);
 341 }
 342
 343 /*
 344  * this forces ioqs to detect contention and yield,
 345  * leading to quicker shutdown
 346  */
 347 void mog_svc_dev_quit_prepare(struct mog_svc *svc)
 348 {
 349         svc->thr_per_dev = 1;
 350         mog_svc_dev_rescale_all(svc);
 351 }