pipe: Add notification lossage handling

Add handling for loss of notifications by having read() insert a
loss-notification message after it has read the pipe buffer that was last
in the ring when the loss occurred.

Lossage can come about either by running out of notification descriptors or
by running out of space in the pipe ring.

Signed-off-by: David Howells <dhowells@redhat.com>

diff --git a/fs/pipe.c b/fs/pipe.c
index 0f9fd897ceb562b8670711f2acc81ba0b1199672..620a113f92eb95cd072fa96ac8b741d0a0ff1fa1 100644 (file)
--- a/fs/pipe.c
+++ b/fs/pipe.c
@@ -314,6 +314,30 @@ pipe_read(struct kiocb *iocb, struct iov_iter *to)
                unsigned int tail = pipe->tail;
                unsigned int mask = pipe->ring_size - 1;
 
+#ifdef CONFIG_WATCH_QUEUE
+               if (pipe->note_loss) {
+                       struct watch_notification n;
+
+                       if (total_len < 8) {
+                               if (ret == 0)
+                                       ret = -ENOBUFS;
+                               break;
+                       }
+
+                       n.type = WATCH_TYPE_META;
+                       n.subtype = WATCH_META_LOSS_NOTIFICATION;
+                       n.info = watch_sizeof(n);
+                       if (copy_to_iter(&n, sizeof(n), to) != sizeof(n)) {
+                               if (ret == 0)
+                                       ret = -EFAULT;
+                               break;
+                       }
+                       ret += sizeof(n);
+                       total_len -= sizeof(n);
+                       pipe->note_loss = false;
+               }
+#endif
+
                if (!pipe_empty(head, tail)) {
                        struct pipe_buffer *buf = &pipe->bufs[tail & mask];
                        size_t chars = buf->len;
@@ -355,6 +379,10 @@ pipe_read(struct kiocb *iocb, struct iov_iter *to)
                        if (!buf->len) {
                                pipe_buf_release(pipe, buf);
                                spin_lock_irq(&pipe->rd_wait.lock);
+#ifdef CONFIG_WATCH_QUEUE
+                               if (buf->flags & PIPE_BUF_FLAG_LOSS)
+                                       pipe->note_loss = true;
+#endif
                                tail++;
                                pipe->tail = tail;
                                spin_unlock_irq(&pipe->rd_wait.lock);

diff --git a/include/linux/pipe_fs_i.h b/include/linux/pipe_fs_i.h
index eaff59a2f07433868045aae74a12cacc21490628..6626f511de6fbade43d5eebf77fc0256c10236c0 100644 (file)
--- a/include/linux/pipe_fs_i.h
+++ b/include/linux/pipe_fs_i.h
@@ -9,6 +9,9 @@
 #define PIPE_BUF_FLAG_GIFT     0x04    /* page is a gift */
 #define PIPE_BUF_FLAG_PACKET   0x08    /* read() as a packet */
 #define PIPE_BUF_FLAG_WHOLE    0x10    /* read() must return entire buffer or error */
+#ifdef CONFIG_WATCH_QUEUE
+#define PIPE_BUF_FLAG_LOSS     0x20    /* Message loss happened after this buffer */
+#endif
 
 /**
  *     struct pipe_buffer - a linux kernel pipe buffer
@@ -34,6 +37,7 @@ struct pipe_buffer {
  *     @wr_wait: writer wait point in case of full pipe
  *     @head: The point of buffer production
  *     @tail: The point of buffer consumption
+ *     @note_loss: The next read() should insert a data-lost message
  *     @max_usage: The maximum number of slots that may be used in the ring
  *     @ring_size: total number of buffers (should be a power of 2)
  *     @nr_accounted: The amount this pipe accounts for in user->pipe_bufs
@@ -56,6 +60,9 @@ struct pipe_inode_info {
        unsigned int tail;
        unsigned int max_usage;
        unsigned int ring_size;
+#ifdef CONFIG_WATCH_QUEUE
+       bool note_loss;
+#endif
        unsigned int nr_accounted;
        unsigned int readers;
        unsigned int writers;

diff --git a/kernel/watch_queue.c b/kernel/watch_queue.c
index ad64ea300f6df37b7045aef9f51ee2bb02779143..9a9699c06709b19602478f4a6754bb6ad8dfc496 100644 (file)
--- a/kernel/watch_queue.c
+++ b/kernel/watch_queue.c
@@ -132,6 +132,8 @@ out:
        return done;
 
 lost:
+       buf = &pipe->bufs[(head - 1) & mask];
+       buf->flags |= PIPE_BUF_FLAG_LOSS;
        goto out;
 }
 

diff --git a/samples/watch_queue/watch_test.c b/samples/watch_queue/watch_test.c
index 8628b4c5d567655a7b4280245c0cb7b8c4f47386..46e618a897fef943c52749f46d8038459ca3f53a 100644 (file)
--- a/samples/watch_queue/watch_test.c
+++ b/samples/watch_queue/watch_test.c
@@ -120,6 +120,9 @@ static void consumer(int fd)
                                               (n.n.info & WATCH_INFO_ID) >>
                                               WATCH_INFO_ID__SHIFT);
                                        break;
+                               case WATCH_META_LOSS_NOTIFICATION:
+                                       printf("-- LOSS --\n");
+                                       break;
                                default:
                                        printf("other meta record\n");
                                        break;