sys/kern/dsched/fq/fq_core.c

   1 /*
   2  * Copyright (c) 2009, 2010 The DragonFly Project.  All rights reserved.
   3  *
   4  * This code is derived from software contributed to The DragonFly Project
   5  * by Alex Hornung <ahornung@gmail.com>
   6  *
   7  * Redistribution and use in source and binary forms, with or without
   8  * modification, are permitted provided that the following conditions
   9  * are met:
  10  *
  11  * 1. Redistributions of source code must retain the above copyright
  12  *    notice, this list of conditions and the following disclaimer.
  13  * 2. Redistributions in binary form must reproduce the above copyright
  14  *    notice, this list of conditions and the following disclaimer in
  15  *    the documentation and/or other materials provided with the
  16  *    distribution.
  17  * 3. Neither the name of The DragonFly Project nor the names of its
  18  *    contributors may be used to endorse or promote products derived
  19  *    from this software without specific, prior written permission.
  20  *
  21  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  22  * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  23  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
  24  * FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE
  25  * COPYRIGHT HOLDERS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
  26  * INCIDENTAL, SPECIAL, EXEMPLARY OR CONSEQUENTIAL DAMAGES (INCLUDING,
  27  * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
  28  * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
  29  * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
  30  * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
  31  * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  32  * SUCH DAMAGE.
  33  */
  34 #include <sys/param.h>
  35 #include <sys/systm.h>
  36 #include <sys/kernel.h>
  37 #include <sys/proc.h>
  38 #include <sys/sysctl.h>
  39 #include <sys/buf.h>
  40 #include <sys/conf.h>
  41 #include <sys/diskslice.h>
  42 #include <sys/disk.h>
  43 #include <machine/atomic.h>
  44 #include <sys/thread.h>
  45 #include <sys/thread2.h>
  46 #include <sys/ctype.h>
  47 #include <sys/buf2.h>
  48 #include <sys/syslog.h>
  49 #include <sys/dsched.h>
  50 #include <machine/param.h>
  51
  52 #include <kern/dsched/fq/fq.h>
  53
  54 static int      dsched_fq_version_maj = 1;
  55 static int      dsched_fq_version_min = 0;
  56
  57 /* Make sure our structs fit */
  58 CTASSERT(sizeof(struct fq_thread_io) <= DSCHED_THREAD_IO_MAX_SZ);
  59 CTASSERT(sizeof(struct fq_disk_ctx) <= DSCHED_DISK_CTX_MAX_SZ);
  60
  61 struct dsched_fq_stats  fq_stats;
  62
  63 extern struct dsched_policy dsched_fq_policy;
  64
  65 void
  66 fq_dispatcher(struct fq_disk_ctx *diskctx)
  67 {
  68         struct dsched_thread_ctx *tdctx;
  69         struct dsched_thread_io *ds_tdio, *ds_tdio2;
  70         struct fq_thread_io     *tdio;
  71         struct bio *bio, *bio2;
  72         int idle;
  73
  74         /*
  75          * We need to manually assign an tdio to the tdctx of this thread
  76          * since it isn't assigned one during fq_prepare, as the disk
  77          * is not set up yet.
  78          */
  79         tdctx = dsched_get_thread_priv(curthread);
  80         KKASSERT(tdctx != NULL);
  81
  82         tdio = (struct fq_thread_io *)dsched_thread_io_alloc(diskctx->head.dp, tdctx, &dsched_fq_policy);
  83
  84         DSCHED_DISK_CTX_LOCK(&diskctx->head);
  85         for(;;) {
  86                 idle = 0;
  87                 /* sleep ~60 ms */
  88                 if ((lksleep(diskctx, &diskctx->head.lock, 0, "fq_dispatcher", hz/15) == 0)) {
  89                         /*
  90                          * We've been woken up; this either means that we are
  91                          * supposed to die away nicely or that the disk is idle.
  92                          */
  93
  94                         if (__predict_false(diskctx->die == 1)) {
  95                                 /* If we are supposed to die, drain all queues */
  96                                 fq_drain(diskctx, FQ_DRAIN_FLUSH);
  97
  98                                 /* Now we can safely unlock and exit */
  99                                 DSCHED_DISK_CTX_UNLOCK(&diskctx->head);
 100                                 kprintf("fq_dispatcher is peacefully dying\n");
 101                                 lwkt_exit();
 102                                 /* NOTREACHED */
 103                         }
 104
 105                         /*
 106                          * We have been awakened because the disk is idle.
 107                          * So let's get ready to dispatch some extra bios.
 108                          */
 109                         idle = 1;
 110                 }
 111
 112                 /* Maybe the disk is idle and we just didn't get the wakeup */
 113                 if (idle == 0)
 114                         idle = diskctx->idle;
 115
 116                 /*
 117                  * XXX: further room for improvements here. It would be better
 118                  *      to dispatch a few requests from each tdio as to ensure
 119                  *      real fairness.
 120                  */
 121                 TAILQ_FOREACH_MUTABLE(ds_tdio, &diskctx->head.tdio_list, dlink, ds_tdio2) {
 122                         tdio = (struct fq_thread_io *)ds_tdio;
 123                         if (tdio->head.qlength == 0)
 124                                 continue;
 125
 126                         DSCHED_THREAD_IO_LOCK(&tdio->head);
 127                         if (atomic_cmpset_int(&tdio->rebalance, 1, 0))
 128                                 fq_balance_self(tdio);
 129                         /*
 130                          * XXX: why 5 extra? should probably be dynamic,
 131                          *      relying on information on latency.
 132                          */
 133                         if ((tdio->max_tp > 0) && idle &&
 134                             (tdio->issued >= tdio->max_tp)) {
 135                                 tdio->max_tp += 5;
 136                         }
 137
 138                         TAILQ_FOREACH_MUTABLE(bio, &tdio->head.queue, link, bio2) {
 139                                 if (atomic_cmpset_int(&tdio->rebalance, 1, 0))
 140                                         fq_balance_self(tdio);
 141                                 if ((tdio->max_tp > 0) &&
 142                                     ((tdio->issued >= tdio->max_tp)))
 143                                         break;
 144
 145                                 TAILQ_REMOVE(&tdio->head.queue, bio, link);
 146                                 --tdio->head.qlength;
 147
 148                                 /*
 149                                  * beware that we do have an tdio reference
 150                                  * from the queueing
 151                                  */
 152                                 fq_dispatch(diskctx, bio, tdio);
 153                         }
 154                         DSCHED_THREAD_IO_UNLOCK(&tdio->head);
 155
 156                 }
 157         }
 158 }
 159
 160 void
 161 fq_balance_thread(struct fq_disk_ctx *diskctx)
 162 {
 163         struct dsched_thread_io *ds_tdio;
 164         struct  fq_thread_io    *tdio;
 165         struct timeval tv, old_tv;
 166         int64_t total_budget, product;
 167         int64_t budget[FQ_PRIO_MAX+1];
 168         int     n, i, sum, total_disk_time;
 169         int     lost_bits;
 170
 171         DSCHED_DISK_CTX_LOCK(&diskctx->head);
 172
 173         getmicrotime(&diskctx->start_interval);
 174
 175         for (;;) {
 176                 /* sleep ~1s */
 177                 if ((lksleep(curthread, &diskctx->head.lock, 0, "fq_balancer", hz/2) == 0)) {
 178                         if (__predict_false(diskctx->die)) {
 179                                 DSCHED_DISK_CTX_UNLOCK(&diskctx->head);
 180                                 lwkt_exit();
 181                         }
 182                 }
 183
 184                 bzero(budget, sizeof(budget));
 185                 total_budget = 0;
 186                 n = 0;
 187
 188                 old_tv = diskctx->start_interval;
 189                 getmicrotime(&tv);
 190
 191                 total_disk_time = (int)(1000000*((tv.tv_sec - old_tv.tv_sec)) +
 192                     (tv.tv_usec - old_tv.tv_usec));
 193
 194                 if (total_disk_time == 0)
 195                         total_disk_time = 1;
 196
 197                 dsched_debug(LOG_INFO, "total_disk_time = %d\n", total_disk_time);
 198
 199                 diskctx->start_interval = tv;
 200
 201                 diskctx->disk_busy = (100*(total_disk_time - diskctx->idle_time)) / total_disk_time;
 202                 if (diskctx->disk_busy < 0)
 203                         diskctx->disk_busy = 0;
 204
 205                 diskctx->idle_time = 0;
 206                 lost_bits = 0;
 207
 208                 TAILQ_FOREACH(ds_tdio, &diskctx->head.tdio_list, dlink) {
 209                         tdio = (struct fq_thread_io *)ds_tdio;
 210                         tdio->interval_avg_latency = tdio->avg_latency;
 211                         tdio->interval_transactions = tdio->transactions;
 212                         if (tdio->interval_transactions > 0) {
 213                                 product = (int64_t)tdio->interval_avg_latency *
 214                                     tdio->interval_transactions;
 215                                 product >>= lost_bits;
 216                                 while(total_budget >= INT64_MAX - product) {
 217                                         ++lost_bits;
 218                                         product >>= 1;
 219                                         total_budget >>= 1;
 220                                 }
 221                                 total_budget += product;
 222                                 ++budget[(tdio->head.p) ? tdio->head.p->p_ionice : 0];
 223                                 KKASSERT(total_budget >= 0);
 224                                 dsched_debug(LOG_INFO,
 225                                     "%d) avg_latency = %d, transactions = %d, ioprio = %d\n",
 226                                     n, tdio->interval_avg_latency, tdio->interval_transactions,
 227                                     (tdio->head.p) ? tdio->head.p->p_ionice : 0);
 228                                 ++n;
 229                         } else {
 230                                 tdio->max_tp = 0;
 231                         }
 232                         tdio->rebalance = 0;
 233                         tdio->transactions = 0;
 234                         tdio->avg_latency = 0;
 235                         tdio->issued = 0;
 236                 }
 237
 238                 dsched_debug(LOG_INFO, "%d procs competing for disk\n"
 239                     "total_budget = %jd (lost bits = %d)\n"
 240                     "incomplete tp = %d\n", n, (intmax_t)total_budget,
 241                     lost_bits, diskctx->incomplete_tp);
 242
 243                 if (n == 0)
 244                         continue;
 245
 246                 sum = 0;
 247
 248                 for (i = 0; i < FQ_PRIO_MAX+1; i++) {
 249                         if (budget[i] == 0)
 250                                 continue;
 251                         sum += (FQ_PRIO_BIAS+i)*budget[i];
 252                 }
 253
 254                 if (sum == 0)
 255                         sum = 1;
 256
 257                 dsched_debug(LOG_INFO, "sum = %d\n", sum);
 258
 259                 for (i = 0; i < FQ_PRIO_MAX+1; i++) {
 260                         if (budget[i] == 0)
 261                                 continue;
 262
 263                         /*
 264                          * XXX: if we still overflow here, we really need to switch to
 265                          *      some more advanced mechanism such as compound int128 or
 266                          *      storing the lost bits so they can be used in the
 267                          *      fq_balance_self.
 268                          */
 269                         diskctx->budgetpb[i] = ((FQ_PRIO_BIAS+i)*total_budget/sum) << lost_bits;
 270                         KKASSERT(diskctx->budgetpb[i] >= 0);
 271                 }
 272
 273                 dsched_debug(4, "disk is %d%% busy\n", diskctx->disk_busy);
 274                 TAILQ_FOREACH(ds_tdio, &diskctx->head.tdio_list, dlink) {
 275                         tdio = (struct fq_thread_io *)ds_tdio;
 276                         tdio->rebalance = 1;
 277                 }
 278
 279                 diskctx->prev_full = diskctx->last_full;
 280                 diskctx->last_full = (diskctx->disk_busy >= 90)?1:0;
 281         }
 282 }
 283
 284
 285 /*
 286  * fq_balance_self should be called from all sorts of dispatchers. It basically
 287  * offloads some of the heavier calculations on throttling onto the process that
 288  * wants to do I/O instead of doing it in the fq_balance thread.
 289  * - should be called with diskctx lock held
 290  */
 291 void
 292 fq_balance_self(struct fq_thread_io *tdio) {
 293         struct fq_disk_ctx *diskctx;
 294
 295         int64_t budget, used_budget;
 296         int64_t avg_latency;
 297         int64_t transactions;
 298
 299         transactions = (int64_t)tdio->interval_transactions;
 300         avg_latency = (int64_t)tdio->interval_avg_latency;
 301         diskctx = (struct fq_disk_ctx *)tdio->head.diskctx;
 302
 303 #if 0
 304         /* XXX: do we really require the lock? */
 305         DSCHED_DISK_CTX_LOCK_ASSERT(diskctx);
 306 #endif
 307
 308         used_budget = ((int64_t)avg_latency * transactions);
 309         budget = diskctx->budgetpb[(tdio->head.p) ? tdio->head.p->p_ionice : 0];
 310
 311         if (used_budget > 0) {
 312                 dsched_debug(LOG_INFO,
 313                     "info: used_budget = %jd, budget = %jd\n",
 314                     (intmax_t)used_budget, budget);
 315         }
 316
 317         if ((used_budget > budget) && (diskctx->disk_busy >= 90)) {
 318                 KKASSERT(avg_latency != 0);
 319
 320                 tdio->max_tp = budget/(avg_latency);
 321                 atomic_add_int(&fq_stats.procs_limited, 1);
 322
 323                 dsched_debug(LOG_INFO,
 324                     "rate limited to %d transactions\n", tdio->max_tp);
 325
 326         } else if (((used_budget*2 < budget) || (diskctx->disk_busy < 80)) &&
 327             (!diskctx->prev_full && !diskctx->last_full)) {
 328                 tdio->max_tp = 0;
 329         }
 330 }
 331
 332
 333 static int
 334 do_fqstats(SYSCTL_HANDLER_ARGS)
 335 {
 336         return (sysctl_handle_opaque(oidp, &fq_stats, sizeof(struct dsched_fq_stats), req));
 337 }
 338
 339 static int
 340 fq_mod_handler(module_t mod, int type, void *unused)
 341 {
 342         static struct sysctl_ctx_list sysctl_ctx;
 343         static struct sysctl_oid *oid;
 344         static char version[16];
 345         int error;
 346
 347         ksnprintf(version, sizeof(version), "%d.%d",
 348             dsched_fq_version_maj, dsched_fq_version_min);
 349
 350         switch (type) {
 351         case MOD_LOAD:
 352                 bzero(&fq_stats, sizeof(struct dsched_fq_stats));
 353                 if ((error = dsched_register(&dsched_fq_policy)))
 354                         return (error);
 355
 356                 sysctl_ctx_init(&sysctl_ctx);
 357                 oid = SYSCTL_ADD_NODE(&sysctl_ctx,
 358                     SYSCTL_STATIC_CHILDREN(_dsched),
 359                     OID_AUTO,
 360                     "fq",
 361                     CTLFLAG_RD, 0, "");
 362
 363                 SYSCTL_ADD_PROC(&sysctl_ctx, SYSCTL_CHILDREN(oid),
 364                     OID_AUTO, "stats", CTLTYPE_OPAQUE|CTLFLAG_RD,
 365                     0, 0, do_fqstats, "S,dsched_fq_stats", "fq statistics");
 366
 367                 SYSCTL_ADD_STRING(&sysctl_ctx, SYSCTL_CHILDREN(oid),
 368                     OID_AUTO, "version", CTLFLAG_RD, version, 0, "fq version");
 369
 370                 kprintf("FQ scheduler policy version %d.%d loaded\n",
 371                     dsched_fq_version_maj, dsched_fq_version_min);
 372                 break;
 373
 374         case MOD_UNLOAD:
 375                 if ((error = dsched_unregister(&dsched_fq_policy)))
 376                         return (error);
 377                 sysctl_ctx_free(&sysctl_ctx);
 378                 kprintf("FQ scheduler policy unloaded\n");
 379                 break;
 380
 381         default:
 382                 break;
 383         }
 384
 385         return 0;
 386 }
 387
 388 DSCHED_POLICY_MODULE(dsched_fq, fq_mod_handler);