drivers/hv/ring_buffer.c

   1 /*
   2  *
   3  * Copyright (c) 2009, Microsoft Corporation.
   4  *
   5  * This program is free software; you can redistribute it and/or modify it
   6  * under the terms and conditions of the GNU General Public License,
   7  * version 2, as published by the Free Software Foundation.
   8  *
   9  * This program is distributed in the hope it will be useful, but WITHOUT
  10  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  11  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
  12  * more details.
  13  *
  14  * You should have received a copy of the GNU General Public License along with
  15  * this program; if not, write to the Free Software Foundation, Inc., 59 Temple
  16  * Place - Suite 330, Boston, MA 02111-1307 USA.
  17  *
  18  * Authors:
  19  *   Haiyang Zhang <haiyangz@microsoft.com>
  20  *   Hank Janssen  <hjanssen@microsoft.com>
  21  *   K. Y. Srinivasan <kys@microsoft.com>
  22  *
  23  */
  24 #define pr_fmt(fmt) KBUILD_MODNAME ": " fmt
  25
  26 #include <linux/kernel.h>
  27 #include <linux/mm.h>
  28 #include <linux/hyperv.h>
  29 #include <linux/uio.h>
  30
  31 #include "hyperv_vmbus.h"
  32
  33 void hv_begin_read(struct hv_ring_buffer_info *rbi)
  34 {
  35         rbi->ring_buffer->interrupt_mask = 1;
  36         mb();
  37 }
  38
  39 u32 hv_end_read(struct hv_ring_buffer_info *rbi)
  40 {
  41         u32 read;
  42         u32 write;
  43
  44         rbi->ring_buffer->interrupt_mask = 0;
  45         mb();
  46
  47         /*
  48          * Now check to see if the ring buffer is still empty.
  49          * If it is not, we raced and we need to process new
  50          * incoming messages.
  51          */
  52         hv_get_ringbuffer_availbytes(rbi, &read, &write);
  53
  54         return read;
  55 }
  56
  57 /*
  58  * When we write to the ring buffer, check if the host needs to
  59  * be signaled. Here is the details of this protocol:
  60  *
  61  *      1. The host guarantees that while it is draining the
  62  *         ring buffer, it will set the interrupt_mask to
  63  *         indicate it does not need to be interrupted when
  64  *         new data is placed.
  65  *
  66  *      2. The host guarantees that it will completely drain
  67  *         the ring buffer before exiting the read loop. Further,
  68  *         once the ring buffer is empty, it will clear the
  69  *         interrupt_mask and re-check to see if new data has
  70  *         arrived.
  71  */
  72
  73 static bool hv_need_to_signal(u32 old_write, struct hv_ring_buffer_info *rbi)
  74 {
  75         mb();
  76         if (rbi->ring_buffer->interrupt_mask)
  77                 return false;
  78
  79         /* check interrupt_mask before read_index */
  80         rmb();
  81         /*
  82          * This is the only case we need to signal when the
  83          * ring transitions from being empty to non-empty.
  84          */
  85         if (old_write == rbi->ring_buffer->read_index)
  86                 return true;
  87
  88         return false;
  89 }
  90
  91 /*
  92  * To optimize the flow management on the send-side,
  93  * when the sender is blocked because of lack of
  94  * sufficient space in the ring buffer, potential the
  95  * consumer of the ring buffer can signal the producer.
  96  * This is controlled by the following parameters:
  97  *
  98  * 1. pending_send_sz: This is the size in bytes that the
  99  *    producer is trying to send.
 100  * 2. The feature bit feat_pending_send_sz set to indicate if
 101  *    the consumer of the ring will signal when the ring
 102  *    state transitions from being full to a state where
 103  *    there is room for the producer to send the pending packet.
 104  */
 105
 106 static bool hv_need_to_signal_on_read(u32 prev_write_sz,
 107                                       struct hv_ring_buffer_info *rbi)
 108 {
 109         u32 cur_write_sz;
 110         u32 r_size;
 111         u32 write_loc = rbi->ring_buffer->write_index;
 112         u32 read_loc = rbi->ring_buffer->read_index;
 113         u32 pending_sz = rbi->ring_buffer->pending_send_sz;
 114
 115         /* If the other end is not blocked on write don't bother. */
 116         if (pending_sz == 0)
 117                 return false;
 118
 119         r_size = rbi->ring_datasize;
 120         cur_write_sz = write_loc >= read_loc ? r_size - (write_loc - read_loc) :
 121                         read_loc - write_loc;
 122
 123         if ((prev_write_sz < pending_sz) && (cur_write_sz >= pending_sz))
 124                 return true;
 125
 126         return false;
 127 }
 128
 129 /* Get the next write location for the specified ring buffer. */
 130 static inline u32
 131 hv_get_next_write_location(struct hv_ring_buffer_info *ring_info)
 132 {
 133         u32 next = ring_info->ring_buffer->write_index;
 134
 135         return next;
 136 }
 137
 138 /* Set the next write location for the specified ring buffer. */
 139 static inline void
 140 hv_set_next_write_location(struct hv_ring_buffer_info *ring_info,
 141                      u32 next_write_location)
 142 {
 143         ring_info->ring_buffer->write_index = next_write_location;
 144 }
 145
 146 /* Get the next read location for the specified ring buffer. */
 147 static inline u32
 148 hv_get_next_read_location(struct hv_ring_buffer_info *ring_info)
 149 {
 150         u32 next = ring_info->ring_buffer->read_index;
 151
 152         return next;
 153 }
 154
 155 /*
 156  * Get the next read location + offset for the specified ring buffer.
 157  * This allows the caller to skip.
 158  */
 159 static inline u32
 160 hv_get_next_readlocation_withoffset(struct hv_ring_buffer_info *ring_info,
 161                                  u32 offset)
 162 {
 163         u32 next = ring_info->ring_buffer->read_index;
 164
 165         next += offset;
 166         next %= ring_info->ring_datasize;
 167
 168         return next;
 169 }
 170
 171 /* Set the next read location for the specified ring buffer. */
 172 static inline void
 173 hv_set_next_read_location(struct hv_ring_buffer_info *ring_info,
 174                     u32 next_read_location)
 175 {
 176         ring_info->ring_buffer->read_index = next_read_location;
 177 }
 178
 179
 180 /* Get the start of the ring buffer. */
 181 static inline void *
 182 hv_get_ring_buffer(struct hv_ring_buffer_info *ring_info)
 183 {
 184         return (void *)ring_info->ring_buffer->buffer;
 185 }
 186
 187
 188 /* Get the size of the ring buffer. */
 189 static inline u32
 190 hv_get_ring_buffersize(struct hv_ring_buffer_info *ring_info)
 191 {
 192         return ring_info->ring_datasize;
 193 }
 194
 195 /* Get the read and write indices as u64 of the specified ring buffer. */
 196 static inline u64
 197 hv_get_ring_bufferindices(struct hv_ring_buffer_info *ring_info)
 198 {
 199         return (u64)ring_info->ring_buffer->write_index << 32;
 200 }
 201
 202 /*
 203  * Helper routine to copy to source from ring buffer.
 204  * Assume there is enough room. Handles wrap-around in src case only!!
 205  */
 206 static u32 hv_copyfrom_ringbuffer(
 207         struct hv_ring_buffer_info      *ring_info,
 208         void                            *dest,
 209         u32                             destlen,
 210         u32                             start_read_offset)
 211 {
 212         void *ring_buffer = hv_get_ring_buffer(ring_info);
 213         u32 ring_buffer_size = hv_get_ring_buffersize(ring_info);
 214
 215         u32 frag_len;
 216
 217         /* wrap-around detected at the src */
 218         if (destlen > ring_buffer_size - start_read_offset) {
 219                 frag_len = ring_buffer_size - start_read_offset;
 220
 221                 memcpy(dest, ring_buffer + start_read_offset, frag_len);
 222                 memcpy(dest + frag_len, ring_buffer, destlen - frag_len);
 223         } else
 224
 225                 memcpy(dest, ring_buffer + start_read_offset, destlen);
 226
 227
 228         start_read_offset += destlen;
 229         start_read_offset %= ring_buffer_size;
 230
 231         return start_read_offset;
 232 }
 233
 234
 235 /*
 236  * Helper routine to copy from source to ring buffer.
 237  * Assume there is enough room. Handles wrap-around in dest case only!!
 238  */
 239 static u32 hv_copyto_ringbuffer(
 240         struct hv_ring_buffer_info      *ring_info,
 241         u32                             start_write_offset,
 242         void                            *src,
 243         u32                             srclen)
 244 {
 245         void *ring_buffer = hv_get_ring_buffer(ring_info);
 246         u32 ring_buffer_size = hv_get_ring_buffersize(ring_info);
 247         u32 frag_len;
 248
 249         /* wrap-around detected! */
 250         if (srclen > ring_buffer_size - start_write_offset) {
 251                 frag_len = ring_buffer_size - start_write_offset;
 252                 memcpy(ring_buffer + start_write_offset, src, frag_len);
 253                 memcpy(ring_buffer, src + frag_len, srclen - frag_len);
 254         } else
 255                 memcpy(ring_buffer + start_write_offset, src, srclen);
 256
 257         start_write_offset += srclen;
 258         start_write_offset %= ring_buffer_size;
 259
 260         return start_write_offset;
 261 }
 262
 263 /* Get various debug metrics for the specified ring buffer. */
 264 void hv_ringbuffer_get_debuginfo(struct hv_ring_buffer_info *ring_info,
 265                             struct hv_ring_buffer_debug_info *debug_info)
 266 {
 267         u32 bytes_avail_towrite;
 268         u32 bytes_avail_toread;
 269
 270         if (ring_info->ring_buffer) {
 271                 hv_get_ringbuffer_availbytes(ring_info,
 272                                         &bytes_avail_toread,
 273                                         &bytes_avail_towrite);
 274
 275                 debug_info->bytes_avail_toread = bytes_avail_toread;
 276                 debug_info->bytes_avail_towrite = bytes_avail_towrite;
 277                 debug_info->current_read_index =
 278                         ring_info->ring_buffer->read_index;
 279                 debug_info->current_write_index =
 280                         ring_info->ring_buffer->write_index;
 281                 debug_info->current_interrupt_mask =
 282                         ring_info->ring_buffer->interrupt_mask;
 283         }
 284 }
 285
 286 /* Initialize the ring buffer. */
 287 int hv_ringbuffer_init(struct hv_ring_buffer_info *ring_info,
 288                    void *buffer, u32 buflen)
 289 {
 290         if (sizeof(struct hv_ring_buffer) != PAGE_SIZE)
 291                 return -EINVAL;
 292
 293         memset(ring_info, 0, sizeof(struct hv_ring_buffer_info));
 294
 295         ring_info->ring_buffer = (struct hv_ring_buffer *)buffer;
 296         ring_info->ring_buffer->read_index =
 297                 ring_info->ring_buffer->write_index = 0;
 298
 299         /* Set the feature bit for enabling flow control. */
 300         ring_info->ring_buffer->feature_bits.value = 1;
 301
 302         ring_info->ring_size = buflen;
 303         ring_info->ring_datasize = buflen - sizeof(struct hv_ring_buffer);
 304
 305         spin_lock_init(&ring_info->ring_lock);
 306
 307         return 0;
 308 }
 309
 310 /* Cleanup the ring buffer. */
 311 void hv_ringbuffer_cleanup(struct hv_ring_buffer_info *ring_info)
 312 {
 313 }
 314
 315 /* Write to the ring buffer. */
 316 int hv_ringbuffer_write(struct hv_ring_buffer_info *outring_info,
 317                     struct kvec *kv_list, u32 kv_count, bool *signal, bool lock)
 318 {
 319         int i = 0;
 320         u32 bytes_avail_towrite;
 321         u32 bytes_avail_toread;
 322         u32 totalbytes_towrite = 0;
 323
 324         u32 next_write_location;
 325         u32 old_write;
 326         u64 prev_indices = 0;
 327         unsigned long flags = 0;
 328
 329         for (i = 0; i < kv_count; i++)
 330                 totalbytes_towrite += kv_list[i].iov_len;
 331
 332         totalbytes_towrite += sizeof(u64);
 333
 334         if (lock)
 335                 spin_lock_irqsave(&outring_info->ring_lock, flags);
 336
 337         hv_get_ringbuffer_availbytes(outring_info,
 338                                 &bytes_avail_toread,
 339                                 &bytes_avail_towrite);
 340
 341         /*
 342          * If there is only room for the packet, assume it is full.
 343          * Otherwise, the next time around, we think the ring buffer
 344          * is empty since the read index == write index.
 345          */
 346         if (bytes_avail_towrite <= totalbytes_towrite) {
 347                 if (lock)
 348                         spin_unlock_irqrestore(&outring_info->ring_lock, flags);
 349                 return -EAGAIN;
 350         }
 351
 352         /* Write to the ring buffer */
 353         next_write_location = hv_get_next_write_location(outring_info);
 354
 355         old_write = next_write_location;
 356
 357         for (i = 0; i < kv_count; i++) {
 358                 next_write_location = hv_copyto_ringbuffer(outring_info,
 359                                                      next_write_location,
 360                                                      kv_list[i].iov_base,
 361                                                      kv_list[i].iov_len);
 362         }
 363
 364         /* Set previous packet start */
 365         prev_indices = hv_get_ring_bufferindices(outring_info);
 366
 367         next_write_location = hv_copyto_ringbuffer(outring_info,
 368                                              next_write_location,
 369                                              &prev_indices,
 370                                              sizeof(u64));
 371
 372         /* Issue a full memory barrier before updating the write index */
 373         mb();
 374
 375         /* Now, update the write location */
 376         hv_set_next_write_location(outring_info, next_write_location);
 377
 378
 379         if (lock)
 380                 spin_unlock_irqrestore(&outring_info->ring_lock, flags);
 381
 382         *signal = hv_need_to_signal(old_write, outring_info);
 383         return 0;
 384 }
 385
 386 int hv_ringbuffer_read(struct hv_ring_buffer_info *inring_info,
 387                        void *buffer, u32 buflen, u32 *buffer_actual_len,
 388                        u64 *requestid, bool *signal, bool raw)
 389 {
 390         u32 bytes_avail_towrite;
 391         u32 bytes_avail_toread;
 392         u32 next_read_location = 0;
 393         u64 prev_indices = 0;
 394         struct vmpacket_descriptor desc;
 395         u32 offset;
 396         u32 packetlen;
 397         int ret = 0;
 398
 399         if (buflen <= 0)
 400                 return -EINVAL;
 401
 402
 403         *buffer_actual_len = 0;
 404         *requestid = 0;
 405
 406         hv_get_ringbuffer_availbytes(inring_info,
 407                                 &bytes_avail_toread,
 408                                 &bytes_avail_towrite);
 409
 410         /* Make sure there is something to read */
 411         if (bytes_avail_toread < sizeof(desc)) {
 412                 /*
 413                  * No error is set when there is even no header, drivers are
 414                  * supposed to analyze buffer_actual_len.
 415                  */
 416                 return ret;
 417         }
 418
 419         next_read_location = hv_get_next_read_location(inring_info);
 420         next_read_location = hv_copyfrom_ringbuffer(inring_info, &desc,
 421                                                     sizeof(desc),
 422                                                     next_read_location);
 423
 424         offset = raw ? 0 : (desc.offset8 << 3);
 425         packetlen = (desc.len8 << 3) - offset;
 426         *buffer_actual_len = packetlen;
 427         *requestid = desc.trans_id;
 428
 429         if (bytes_avail_toread < packetlen + offset)
 430                 return -EAGAIN;
 431
 432         if (packetlen > buflen)
 433                 return -ENOBUFS;
 434
 435         next_read_location =
 436                 hv_get_next_readlocation_withoffset(inring_info, offset);
 437
 438         next_read_location = hv_copyfrom_ringbuffer(inring_info,
 439                                                 buffer,
 440                                                 packetlen,
 441                                                 next_read_location);
 442
 443         next_read_location = hv_copyfrom_ringbuffer(inring_info,
 444                                                 &prev_indices,
 445                                                 sizeof(u64),
 446                                                 next_read_location);
 447
 448         /*
 449          * Make sure all reads are done before we update the read index since
 450          * the writer may start writing to the read area once the read index
 451          * is updated.
 452          */
 453         mb();
 454
 455         /* Update the read index */
 456         hv_set_next_read_location(inring_info, next_read_location);
 457
 458         *signal = hv_need_to_signal_on_read(bytes_avail_towrite, inring_info);
 459
 460         return ret;
 461 }