Line data    Source code

       1             : /*   SPDX-License-Identifier: BSD-3-Clause
       2             :  *   Copyright (C) 2019 Intel Corporation.
       3             :  *   All rights reserved.
       4             :  */
       5             : 
       6             : #include "spdk/pipe.h"
       7             : #include "spdk/util.h"
       8             : #include "spdk/queue.h"
       9             : #include "spdk/log.h"
      10             : 
      11             : struct spdk_pipe_buf {
      12             :         SLIST_ENTRY(spdk_pipe_buf)      link;
      13             :         uint32_t                        sz;
      14             : };
      15             : 
      16             : struct spdk_pipe_group {
      17             :         SLIST_HEAD(, spdk_pipe_buf) bufs;
      18             : };
      19             : 
      20             : struct spdk_pipe {
      21             :         uint8_t *buf;
      22             :         uint32_t sz;
      23             : 
      24             :         uint32_t write;
      25             :         uint32_t read;
      26             :         bool full;
      27             : 
      28             :         struct spdk_pipe_group *group;
      29             : };
      30             : 
      31             : struct spdk_pipe *
      32           6 : spdk_pipe_create(void *buf, uint32_t sz)
      33             : {
      34             :         struct spdk_pipe *pipe;
      35             : 
      36           6 :         pipe = calloc(1, sizeof(*pipe));
      37           6 :         if (pipe == NULL) {
      38           0 :                 return NULL;
      39             :         }
      40             : 
      41           6 :         pipe->buf = buf;
      42           6 :         pipe->sz = sz;
      43             : 
      44           6 :         return pipe;
      45           6 : }
      46             : 
      47             : void *
      48           6 : spdk_pipe_destroy(struct spdk_pipe *pipe)
      49             : {
      50             :         void *buf;
      51             : 
      52           6 :         if (pipe == NULL) {
      53           0 :                 return NULL;
      54             :         }
      55             : 
      56           6 :         if (pipe->group) {
      57           0 :                 spdk_pipe_group_remove(pipe->group, pipe);
      58           0 :         }
      59             : 
      60           6 :         buf = pipe->buf;
      61           6 :         free(pipe);
      62           6 :         return buf;
      63           6 : }
      64             : 
      65             : static void
      66           0 : pipe_alloc_buf_from_group(struct spdk_pipe *pipe)
      67             : {
      68             :         struct spdk_pipe_buf *buf;
      69             :         struct spdk_pipe_group *group;
      70             : 
      71           0 :         assert(pipe->group != NULL);
      72           0 :         group = pipe->group;
      73             : 
      74             :         /* We have to pick a buffer that's the correct size. It's almost always
      75             :          * the first one. */
      76           0 :         buf = SLIST_FIRST(&group->bufs);
      77           0 :         while (buf != NULL) {
      78           0 :                 if (buf->sz == pipe->sz) {
      79             :                         /* TODO: Could track the previous and do an SLIST_REMOVE_AFTER */
      80           0 :                         SLIST_REMOVE(&pipe->group->bufs, buf, spdk_pipe_buf, link);
      81           0 :                         pipe->buf = (void *)buf;
      82           0 :                         return;
      83             :                 }
      84           0 :                 buf = SLIST_NEXT(buf, link);
      85             :         }
      86             :         /* Should never get here. */
      87           0 :         assert(false);
      88             : }
      89             : 
      90             : int
      91          13 : spdk_pipe_writer_get_buffer(struct spdk_pipe *pipe, uint32_t requested_sz, struct iovec *iovs)
      92             : {
      93             :         uint32_t sz;
      94             :         uint32_t read;
      95             :         uint32_t write;
      96             : 
      97          13 :         read = pipe->read;
      98          13 :         write = pipe->write;
      99             : 
     100          13 :         if (pipe->full || requested_sz == 0) {
     101           2 :                 iovs[0].iov_base = NULL;
     102           2 :                 iovs[0].iov_len = 0;
     103           2 :                 return 0;
     104             :         }
     105             : 
     106          11 :         if (pipe->buf == NULL) {
     107           0 :                 pipe_alloc_buf_from_group(pipe);
     108           0 :         }
     109             : 
     110          11 :         if (read <= write) {
     111           9 :                 sz = spdk_min(requested_sz, pipe->sz - write);
     112             : 
     113           9 :                 iovs[0].iov_base = pipe->buf + write;
     114           9 :                 iovs[0].iov_len = sz;
     115             : 
     116           9 :                 requested_sz -= sz;
     117             : 
     118           9 :                 if (requested_sz > 0) {
     119           3 :                         sz = spdk_min(requested_sz, read);
     120             : 
     121           3 :                         iovs[1].iov_base = (sz == 0) ? NULL : pipe->buf;
     122           3 :                         iovs[1].iov_len = sz;
     123           3 :                 } else {
     124           6 :                         iovs[1].iov_base = NULL;
     125           6 :                         iovs[1].iov_len = 0;
     126             :                 }
     127           9 :         } else {
     128           2 :                 sz = spdk_min(requested_sz, read - write);
     129             : 
     130           2 :                 iovs[0].iov_base = pipe->buf + write;
     131           2 :                 iovs[0].iov_len = sz;
     132           2 :                 iovs[1].iov_base = NULL;
     133           2 :                 iovs[1].iov_len = 0;
     134             :         }
     135             : 
     136          11 :         return iovs[0].iov_len + iovs[1].iov_len;
     137          13 : }
     138             : 
     139             : int
     140          11 : spdk_pipe_writer_advance(struct spdk_pipe *pipe, uint32_t requested_sz)
     141             : {
     142             :         uint32_t sz;
     143             :         uint32_t read;
     144             :         uint32_t write;
     145             : 
     146          11 :         read = pipe->read;
     147          11 :         write = pipe->write;
     148             : 
     149          11 :         if (requested_sz > pipe->sz || pipe->full) {
     150           1 :                 return -EINVAL;
     151             :         }
     152             : 
     153          10 :         if (read <= write) {
     154           7 :                 if (requested_sz > (pipe->sz - write) + read) {
     155           1 :                         return -EINVAL;
     156             :                 }
     157             : 
     158           6 :                 sz = spdk_min(requested_sz, pipe->sz - write);
     159             : 
     160           6 :                 write += sz;
     161           6 :                 if (write == pipe->sz) {
     162           3 :                         write = 0;
     163           3 :                 }
     164           6 :                 requested_sz -= sz;
     165             : 
     166           6 :                 if (requested_sz > 0) {
     167           1 :                         write = requested_sz;
     168           1 :                 }
     169           6 :         } else {
     170           3 :                 if (requested_sz > (read - write)) {
     171           2 :                         return -EINVAL;
     172             :                 }
     173             : 
     174           1 :                 write += requested_sz;
     175             :         }
     176             : 
     177           7 :         if (read == write) {
     178           3 :                 pipe->full = true;
     179           3 :         }
     180           7 :         pipe->write = write;
     181             : 
     182           7 :         return 0;
     183          11 : }
     184             : 
     185             : uint32_t
     186           3 : spdk_pipe_reader_bytes_available(struct spdk_pipe *pipe)
     187             : {
     188             :         uint32_t read;
     189             :         uint32_t write;
     190             : 
     191           3 :         read = pipe->read;
     192           3 :         write = pipe->write;
     193             : 
     194           3 :         if (read == write && !pipe->full) {
     195           0 :                 return 0;
     196           3 :         } else if (read < write) {
     197           2 :                 return write - read;
     198             :         } else {
     199           1 :                 return (pipe->sz - read) + write;
     200             :         }
     201           3 : }
     202             : 
     203             : int
     204          12 : spdk_pipe_reader_get_buffer(struct spdk_pipe *pipe, uint32_t requested_sz, struct iovec *iovs)
     205             : {
     206             :         uint32_t sz;
     207             :         uint32_t read;
     208             :         uint32_t write;
     209             : 
     210          12 :         read = pipe->read;
     211          12 :         write = pipe->write;
     212             : 
     213          12 :         if ((read == write && !pipe->full) || requested_sz == 0) {
     214           2 :                 iovs[0].iov_base = NULL;
     215           2 :                 iovs[0].iov_len = 0;
     216           2 :                 iovs[1].iov_base = NULL;
     217           2 :                 iovs[1].iov_len = 0;
     218          12 :         } else if (read < write) {
     219           5 :                 sz = spdk_min(requested_sz, write - read);
     220             : 
     221           5 :                 iovs[0].iov_base = pipe->buf + read;
     222           5 :                 iovs[0].iov_len = sz;
     223           5 :                 iovs[1].iov_base = NULL;
     224           5 :                 iovs[1].iov_len = 0;
     225           5 :         } else {
     226           5 :                 sz = spdk_min(requested_sz, pipe->sz - read);
     227             : 
     228           5 :                 iovs[0].iov_base = pipe->buf + read;
     229           5 :                 iovs[0].iov_len = sz;
     230             : 
     231           5 :                 requested_sz -= sz;
     232             : 
     233           5 :                 if (requested_sz > 0) {
     234           3 :                         sz = spdk_min(requested_sz, write);
     235           3 :                         iovs[1].iov_base = (sz == 0) ? NULL : pipe->buf;
     236           3 :                         iovs[1].iov_len = sz;
     237           3 :                 } else {
     238           2 :                         iovs[1].iov_base = NULL;
     239           2 :                         iovs[1].iov_len = 0;
     240             :                 }
     241             :         }
     242             : 
     243          12 :         return iovs[0].iov_len + iovs[1].iov_len;
     244             : }
     245             : 
     246             : int
     247           9 : spdk_pipe_reader_advance(struct spdk_pipe *pipe, uint32_t requested_sz)
     248             : {
     249             :         uint32_t sz;
     250             :         uint32_t read;
     251             :         uint32_t write;
     252             : 
     253           9 :         read = pipe->read;
     254           9 :         write = pipe->write;
     255             : 
     256           9 :         if (requested_sz == 0) {
     257           0 :                 return 0;
     258             :         }
     259             : 
     260           9 :         if (read < write) {
     261           7 :                 if (requested_sz > (write - read)) {
     262           2 :                         return -EINVAL;
     263             :                 }
     264             : 
     265           5 :                 read += requested_sz;
     266           5 :         } else {
     267           2 :                 sz = spdk_min(requested_sz, pipe->sz - read);
     268             : 
     269           2 :                 read += sz;
     270           2 :                 if (read == pipe->sz) {
     271           2 :                         read = 0;
     272           2 :                 }
     273           2 :                 requested_sz -= sz;
     274             : 
     275           2 :                 if (requested_sz > 0) {
     276           1 :                         if (requested_sz > write) {
     277           0 :                                 return -EINVAL;
     278             :                         }
     279             : 
     280           1 :                         read = requested_sz;
     281           1 :                 }
     282             :         }
     283             : 
     284             :         /* We know we advanced at least one byte, so the pipe isn't full. */
     285           7 :         pipe->full = false;
     286             : 
     287           7 :         if (read == write) {
     288             :                 /* The pipe is empty. To re-use the same memory more frequently, jump
     289             :                  * both pointers back to the beginning of the pipe. */
     290           5 :                 read = 0;
     291           5 :                 pipe->write = 0;
     292             : 
     293             :                 /* Additionally, release the buffer to the shared pool */
     294           5 :                 if (pipe->group) {
     295           0 :                         struct spdk_pipe_buf *buf = (struct spdk_pipe_buf *)pipe->buf;
     296           0 :                         buf->sz = pipe->sz;
     297           0 :                         SLIST_INSERT_HEAD(&pipe->group->bufs, buf, link);
     298           0 :                         pipe->buf = NULL;
     299           0 :                 }
     300           5 :         }
     301             : 
     302           7 :         pipe->read = read;
     303             : 
     304           7 :         return 0;
     305           9 : }
     306             : 
     307             : struct spdk_pipe_group *
     308           0 : spdk_pipe_group_create(void)
     309             : {
     310             :         struct spdk_pipe_group *group;
     311             : 
     312           0 :         group = calloc(1, sizeof(*group));
     313           0 :         if (!group) {
     314           0 :                 return NULL;
     315             :         }
     316             : 
     317           0 :         SLIST_INIT(&group->bufs);
     318             : 
     319           0 :         return group;
     320           0 : }
     321             : 
     322             : void
     323           0 : spdk_pipe_group_destroy(struct spdk_pipe_group *group)
     324             : {
     325           0 :         if (!SLIST_EMPTY(&group->bufs)) {
     326           0 :                 SPDK_ERRLOG("Destroying a pipe group that still has buffers!\n");
     327           0 :                 assert(false);
     328             :         }
     329             : 
     330           0 :         free(group);
     331           0 : }
     332             : 
     333             : int
     334           0 : spdk_pipe_group_add(struct spdk_pipe_group *group, struct spdk_pipe *pipe)
     335             : {
     336             :         struct spdk_pipe_buf *buf;
     337             : 
     338           0 :         assert(pipe->group == NULL);
     339             : 
     340           0 :         pipe->group = group;
     341           0 :         if (pipe->read != pipe->write || pipe->full) {
     342             :                 /* Pipe currently has valid data, so keep the buffer attached
     343             :                  * to the pipe for now.  We can move it to the group's SLIST
     344             :                  * later when it gets emptied.
     345             :                  */
     346           0 :                 return 0;
     347             :         }
     348             : 
     349           0 :         buf = (struct spdk_pipe_buf *)pipe->buf;
     350           0 :         buf->sz = pipe->sz;
     351           0 :         SLIST_INSERT_HEAD(&group->bufs, buf, link);
     352           0 :         pipe->buf = NULL;
     353           0 :         return 0;
     354           0 : }
     355             : 
     356             : int
     357           0 : spdk_pipe_group_remove(struct spdk_pipe_group *group, struct spdk_pipe *pipe)
     358             : {
     359           0 :         assert(pipe->group == group);
     360             : 
     361           0 :         if (pipe->buf == NULL) {
     362             :                 /* Associate a buffer with the pipe before returning. */
     363           0 :                 pipe_alloc_buf_from_group(pipe);
     364           0 :                 assert(pipe->buf != NULL);
     365           0 :         }
     366             : 
     367           0 :         pipe->group = NULL;
     368           0 :         return 0;
     369             : }