fs/bcachefs/thread_with_file.c

   1 // SPDX-License-Identifier: GPL-2.0
   2 #ifndef NO_BCACHEFS_FS
   3
   4 #include "bcachefs.h"
   5 #include "thread_with_file.h"
   6
   7 #include <linux/anon_inodes.h>
   8 #include <linux/file.h>
   9 #include <linux/kthread.h>
  10 #include <linux/pagemap.h>
  11 #include <linux/poll.h>
  12 #include <linux/sched/sysctl.h>
  13
  14 void bch2_thread_with_file_exit(struct thread_with_file *thr)
  15 {
  16         if (thr->task) {
  17                 kthread_stop(thr->task);
  18                 put_task_struct(thr->task);
  19         }
  20 }
  21
  22 int bch2_run_thread_with_file(struct thread_with_file *thr,
  23                               const struct file_operations *fops,
  24                               int (*fn)(void *))
  25 {
  26         struct file *file = NULL;
  27         int ret, fd = -1;
  28         unsigned fd_flags = O_CLOEXEC;
  29
  30         if (fops->read && fops->write)
  31                 fd_flags |= O_RDWR;
  32         else if (fops->read)
  33                 fd_flags |= O_RDONLY;
  34         else if (fops->write)
  35                 fd_flags |= O_WRONLY;
  36
  37         char name[TASK_COMM_LEN];
  38         get_task_comm(name, current);
  39
  40         thr->ret = 0;
  41         thr->task = kthread_create(fn, thr, "%s", name);
  42         ret = PTR_ERR_OR_ZERO(thr->task);
  43         if (ret)
  44                 return ret;
  45
  46         ret = get_unused_fd_flags(fd_flags);
  47         if (ret < 0)
  48                 goto err;
  49         fd = ret;
  50
  51         file = anon_inode_getfile(name, fops, thr, fd_flags);
  52         ret = PTR_ERR_OR_ZERO(file);
  53         if (ret)
  54                 goto err;
  55
  56         get_task_struct(thr->task);
  57         wake_up_process(thr->task);
  58         fd_install(fd, file);
  59         return fd;
  60 err:
  61         if (fd >= 0)
  62                 put_unused_fd(fd);
  63         if (thr->task)
  64                 kthread_stop(thr->task);
  65         return ret;
  66 }
  67
  68 /* stdio_redirect */
  69
  70 static bool stdio_redirect_has_more_input(struct stdio_redirect *stdio, size_t seen)
  71 {
  72         return stdio->input.buf.nr > seen || stdio->done;
  73 }
  74
  75 static bool stdio_redirect_has_input(struct stdio_redirect *stdio)
  76 {
  77         return stdio_redirect_has_more_input(stdio, 0);
  78 }
  79
  80 static bool stdio_redirect_has_output(struct stdio_redirect *stdio)
  81 {
  82         return stdio->output.buf.nr || stdio->done;
  83 }
  84
  85 #define STDIO_REDIRECT_BUFSIZE          4096
  86
  87 static bool stdio_redirect_has_input_space(struct stdio_redirect *stdio)
  88 {
  89         return stdio->input.buf.nr < STDIO_REDIRECT_BUFSIZE || stdio->done;
  90 }
  91
  92 static bool stdio_redirect_has_output_space(struct stdio_redirect *stdio)
  93 {
  94         return stdio->output.buf.nr < STDIO_REDIRECT_BUFSIZE || stdio->done;
  95 }
  96
  97 static void stdio_buf_init(struct stdio_buf *buf)
  98 {
  99         spin_lock_init(&buf->lock);
 100         init_waitqueue_head(&buf->wait);
 101         darray_init(&buf->buf);
 102 }
 103
 104 /* thread_with_stdio */
 105
 106 static void thread_with_stdio_done(struct thread_with_stdio *thr)
 107 {
 108         thr->thr.done = true;
 109         thr->stdio.done = true;
 110         wake_up(&thr->stdio.input.wait);
 111         wake_up(&thr->stdio.output.wait);
 112 }
 113
 114 static ssize_t thread_with_stdio_read(struct file *file, char __user *ubuf,
 115                                       size_t len, loff_t *ppos)
 116 {
 117         struct thread_with_stdio *thr =
 118                 container_of(file->private_data, struct thread_with_stdio, thr);
 119         struct stdio_buf *buf = &thr->stdio.output;
 120         size_t copied = 0, b;
 121         int ret = 0;
 122
 123         if (!(file->f_flags & O_NONBLOCK)) {
 124                 ret = wait_event_interruptible(buf->wait, stdio_redirect_has_output(&thr->stdio));
 125                 if (ret)
 126                         return ret;
 127         } else if (!stdio_redirect_has_output(&thr->stdio))
 128                 return -EAGAIN;
 129
 130         while (len && buf->buf.nr) {
 131                 if (fault_in_writeable(ubuf, len) == len) {
 132                         ret = -EFAULT;
 133                         break;
 134                 }
 135
 136                 spin_lock_irq(&buf->lock);
 137                 b = min_t(size_t, len, buf->buf.nr);
 138
 139                 if (b && !copy_to_user_nofault(ubuf, buf->buf.data, b)) {
 140                         ubuf    += b;
 141                         len     -= b;
 142                         copied  += b;
 143                         buf->buf.nr -= b;
 144                         memmove(buf->buf.data,
 145                                 buf->buf.data + b,
 146                                 buf->buf.nr);
 147                 }
 148                 spin_unlock_irq(&buf->lock);
 149         }
 150
 151         return copied ?: ret;
 152 }
 153
 154 static int thread_with_stdio_release(struct inode *inode, struct file *file)
 155 {
 156         struct thread_with_stdio *thr =
 157                 container_of(file->private_data, struct thread_with_stdio, thr);
 158
 159         thread_with_stdio_done(thr);
 160         bch2_thread_with_file_exit(&thr->thr);
 161         darray_exit(&thr->stdio.input.buf);
 162         darray_exit(&thr->stdio.output.buf);
 163         thr->ops->exit(thr);
 164         return 0;
 165 }
 166
 167 static ssize_t thread_with_stdio_write(struct file *file, const char __user *ubuf,
 168                                        size_t len, loff_t *ppos)
 169 {
 170         struct thread_with_stdio *thr =
 171                 container_of(file->private_data, struct thread_with_stdio, thr);
 172         struct stdio_buf *buf = &thr->stdio.input;
 173         size_t copied = 0;
 174         ssize_t ret = 0;
 175
 176         while (len) {
 177                 if (thr->thr.done) {
 178                         ret = -EPIPE;
 179                         break;
 180                 }
 181
 182                 size_t b = len - fault_in_readable(ubuf, len);
 183                 if (!b) {
 184                         ret = -EFAULT;
 185                         break;
 186                 }
 187
 188                 spin_lock(&buf->lock);
 189                 size_t makeroom = b;
 190                 if (!buf->waiting_for_line || memchr(buf->buf.data, '\n', buf->buf.nr))
 191                         makeroom = min_t(ssize_t, makeroom,
 192                                    max_t(ssize_t, STDIO_REDIRECT_BUFSIZE - buf->buf.nr,
 193                                                   0));
 194                 darray_make_room_gfp(&buf->buf, makeroom, GFP_NOWAIT);
 195
 196                 b = min(len, darray_room(buf->buf));
 197
 198                 if (b && !copy_from_user_nofault(&darray_top(buf->buf), ubuf, b)) {
 199                         buf->buf.nr += b;
 200                         ubuf    += b;
 201                         len     -= b;
 202                         copied  += b;
 203                 }
 204                 spin_unlock(&buf->lock);
 205
 206                 if (b) {
 207                         wake_up(&buf->wait);
 208                 } else {
 209                         if ((file->f_flags & O_NONBLOCK)) {
 210                                 ret = -EAGAIN;
 211                                 break;
 212                         }
 213
 214                         ret = wait_event_interruptible(buf->wait,
 215                                         stdio_redirect_has_input_space(&thr->stdio));
 216                         if (ret)
 217                                 break;
 218                 }
 219         }
 220
 221         return copied ?: ret;
 222 }
 223
 224 static __poll_t thread_with_stdio_poll(struct file *file, struct poll_table_struct *wait)
 225 {
 226         struct thread_with_stdio *thr =
 227                 container_of(file->private_data, struct thread_with_stdio, thr);
 228
 229         poll_wait(file, &thr->stdio.output.wait, wait);
 230         poll_wait(file, &thr->stdio.input.wait, wait);
 231
 232         __poll_t mask = 0;
 233
 234         if (stdio_redirect_has_output(&thr->stdio))
 235                 mask |= EPOLLIN;
 236         if (stdio_redirect_has_input_space(&thr->stdio))
 237                 mask |= EPOLLOUT;
 238         if (thr->thr.done)
 239                 mask |= EPOLLHUP|EPOLLERR;
 240         return mask;
 241 }
 242
 243 static __poll_t thread_with_stdout_poll(struct file *file, struct poll_table_struct *wait)
 244 {
 245         struct thread_with_stdio *thr =
 246                 container_of(file->private_data, struct thread_with_stdio, thr);
 247
 248         poll_wait(file, &thr->stdio.output.wait, wait);
 249
 250         __poll_t mask = 0;
 251
 252         if (stdio_redirect_has_output(&thr->stdio))
 253                 mask |= EPOLLIN;
 254         if (thr->thr.done)
 255                 mask |= EPOLLHUP|EPOLLERR;
 256         return mask;
 257 }
 258
 259 static int thread_with_stdio_flush(struct file *file, fl_owner_t id)
 260 {
 261         struct thread_with_stdio *thr =
 262                 container_of(file->private_data, struct thread_with_stdio, thr);
 263
 264         return thr->thr.ret;
 265 }
 266
 267 static long thread_with_stdio_ioctl(struct file *file, unsigned int cmd, unsigned long p)
 268 {
 269         struct thread_with_stdio *thr =
 270                 container_of(file->private_data, struct thread_with_stdio, thr);
 271
 272         if (thr->ops->unlocked_ioctl)
 273                 return thr->ops->unlocked_ioctl(thr, cmd, p);
 274         return -ENOTTY;
 275 }
 276
 277 static const struct file_operations thread_with_stdio_fops = {
 278         .read           = thread_with_stdio_read,
 279         .write          = thread_with_stdio_write,
 280         .poll           = thread_with_stdio_poll,
 281         .flush          = thread_with_stdio_flush,
 282         .release        = thread_with_stdio_release,
 283         .unlocked_ioctl = thread_with_stdio_ioctl,
 284 };
 285
 286 static const struct file_operations thread_with_stdout_fops = {
 287         .read           = thread_with_stdio_read,
 288         .poll           = thread_with_stdout_poll,
 289         .flush          = thread_with_stdio_flush,
 290         .release        = thread_with_stdio_release,
 291         .unlocked_ioctl = thread_with_stdio_ioctl,
 292 };
 293
 294 static int thread_with_stdio_fn(void *arg)
 295 {
 296         struct thread_with_stdio *thr = arg;
 297
 298         thr->thr.ret = thr->ops->fn(thr);
 299
 300         thread_with_stdio_done(thr);
 301         return 0;
 302 }
 303
 304 void bch2_thread_with_stdio_init(struct thread_with_stdio *thr,
 305                                  const struct thread_with_stdio_ops *ops)
 306 {
 307         stdio_buf_init(&thr->stdio.input);
 308         stdio_buf_init(&thr->stdio.output);
 309         thr->ops = ops;
 310 }
 311
 312 int __bch2_run_thread_with_stdio(struct thread_with_stdio *thr)
 313 {
 314         return bch2_run_thread_with_file(&thr->thr, &thread_with_stdio_fops, thread_with_stdio_fn);
 315 }
 316
 317 int bch2_run_thread_with_stdio(struct thread_with_stdio *thr,
 318                                const struct thread_with_stdio_ops *ops)
 319 {
 320         bch2_thread_with_stdio_init(thr, ops);
 321
 322         return __bch2_run_thread_with_stdio(thr);
 323 }
 324
 325 int bch2_run_thread_with_stdout(struct thread_with_stdio *thr,
 326                                 const struct thread_with_stdio_ops *ops)
 327 {
 328         stdio_buf_init(&thr->stdio.input);
 329         stdio_buf_init(&thr->stdio.output);
 330         thr->ops = ops;
 331
 332         return bch2_run_thread_with_file(&thr->thr, &thread_with_stdout_fops, thread_with_stdio_fn);
 333 }
 334 EXPORT_SYMBOL_GPL(bch2_run_thread_with_stdout);
 335
 336 int bch2_stdio_redirect_read(struct stdio_redirect *stdio, char *ubuf, size_t len)
 337 {
 338         struct stdio_buf *buf = &stdio->input;
 339
 340         /*
 341          * we're waiting on user input (or for the file descriptor to be
 342          * closed), don't want a hung task warning:
 343          */
 344         do {
 345                 wait_event_timeout(buf->wait, stdio_redirect_has_input(stdio),
 346                                    sysctl_hung_task_timeout_secs * HZ / 2);
 347         } while (!stdio_redirect_has_input(stdio));
 348
 349         if (stdio->done)
 350                 return -1;
 351
 352         spin_lock(&buf->lock);
 353         int ret = min(len, buf->buf.nr);
 354         buf->buf.nr -= ret;
 355         memcpy(ubuf, buf->buf.data, ret);
 356         memmove(buf->buf.data,
 357                 buf->buf.data + ret,
 358                 buf->buf.nr);
 359         spin_unlock(&buf->lock);
 360
 361         wake_up(&buf->wait);
 362         return ret;
 363 }
 364
 365 int bch2_stdio_redirect_readline_timeout(struct stdio_redirect *stdio,
 366                                          darray_char *line,
 367                                          unsigned long timeout)
 368 {
 369         unsigned long until = jiffies + timeout, t;
 370         struct stdio_buf *buf = &stdio->input;
 371         size_t seen = 0;
 372 again:
 373         t = timeout != MAX_SCHEDULE_TIMEOUT
 374                 ? max_t(long, until - jiffies, 0)
 375                 : timeout;
 376
 377         t = min(t, sysctl_hung_task_timeout_secs * HZ / 2);
 378
 379         wait_event_timeout(buf->wait, stdio_redirect_has_more_input(stdio, seen), t);
 380
 381         if (stdio->done)
 382                 return -1;
 383
 384         spin_lock(&buf->lock);
 385         seen = buf->buf.nr;
 386         char *n = memchr(buf->buf.data, '\n', seen);
 387
 388         if (!n && timeout != MAX_SCHEDULE_TIMEOUT && time_after_eq(jiffies, until)) {
 389                 spin_unlock(&buf->lock);
 390                 return -ETIME;
 391         }
 392
 393         if (!n) {
 394                 buf->waiting_for_line = true;
 395                 spin_unlock(&buf->lock);
 396                 goto again;
 397         }
 398
 399         size_t b = n + 1 - buf->buf.data;
 400         if (b > line->size) {
 401                 spin_unlock(&buf->lock);
 402                 int ret = darray_resize(line, b);
 403                 if (ret)
 404                         return ret;
 405                 seen = 0;
 406                 goto again;
 407         }
 408
 409         buf->buf.nr -= b;
 410         memcpy(line->data, buf->buf.data, b);
 411         memmove(buf->buf.data,
 412                 buf->buf.data + b,
 413                 buf->buf.nr);
 414         line->nr = b;
 415
 416         buf->waiting_for_line = false;
 417         spin_unlock(&buf->lock);
 418
 419         wake_up(&buf->wait);
 420         return 0;
 421 }
 422
 423 int bch2_stdio_redirect_readline(struct stdio_redirect *stdio, darray_char *line)
 424 {
 425         return bch2_stdio_redirect_readline_timeout(stdio, line, MAX_SCHEDULE_TIMEOUT);
 426 }
 427
 428 __printf(3, 0)
 429 static ssize_t bch2_darray_vprintf(darray_char *out, gfp_t gfp, const char *fmt, va_list args)
 430 {
 431         ssize_t ret;
 432
 433         do {
 434                 va_list args2;
 435                 size_t len;
 436
 437                 va_copy(args2, args);
 438                 len = vsnprintf(out->data + out->nr, darray_room(*out), fmt, args2);
 439                 va_end(args2);
 440
 441                 if (len + 1 <= darray_room(*out)) {
 442                         out->nr += len;
 443                         return len;
 444                 }
 445
 446                 ret = darray_make_room_gfp(out, len + 1, gfp);
 447         } while (ret == 0);
 448
 449         return ret;
 450 }
 451
 452 ssize_t bch2_stdio_redirect_vprintf(struct stdio_redirect *stdio, bool nonblocking,
 453                                     const char *fmt, va_list args)
 454 {
 455         struct stdio_buf *buf = &stdio->output;
 456         unsigned long flags;
 457         ssize_t ret;
 458
 459 again:
 460         spin_lock_irqsave(&buf->lock, flags);
 461         ret = bch2_darray_vprintf(&buf->buf, GFP_NOWAIT, fmt, args);
 462         spin_unlock_irqrestore(&buf->lock, flags);
 463
 464         if (ret < 0) {
 465                 if (nonblocking)
 466                         return -EAGAIN;
 467
 468                 ret = wait_event_interruptible(buf->wait,
 469                                 stdio_redirect_has_output_space(stdio));
 470                 if (ret)
 471                         return ret;
 472                 goto again;
 473         }
 474
 475         wake_up(&buf->wait);
 476         return ret;
 477 }
 478
 479 ssize_t bch2_stdio_redirect_printf(struct stdio_redirect *stdio, bool nonblocking,
 480                                 const char *fmt, ...)
 481 {
 482         va_list args;
 483         ssize_t ret;
 484
 485         va_start(args, fmt);
 486         ret = bch2_stdio_redirect_vprintf(stdio, nonblocking, fmt, args);
 487         va_end(args);
 488
 489         return ret;
 490 }
 491
 492 #endif /* NO_BCACHEFS_FS */