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 old_rd,
 107                                          struct hv_ring_buffer_info *rbi)
 108 {
 109         u32 prev_write_sz;
 110         u32 cur_write_sz;
 111         u32 r_size;
 112         u32 write_loc = rbi->ring_buffer->write_index;
 113         u32 read_loc = rbi->ring_buffer->read_index;
 114         u32 pending_sz = rbi->ring_buffer->pending_send_sz;
 115
 116         /*
 117          * If the other end is not blocked on write don't bother.
 118          */
 119         if (pending_sz == 0)
 120                 return false;
 121
 122         r_size = rbi->ring_datasize;
 123         cur_write_sz = write_loc >= read_loc ? r_size - (write_loc - read_loc) :
 124                         read_loc - write_loc;
 125
 126         prev_write_sz = write_loc >= old_rd ? r_size - (write_loc - old_rd) :
 127                         old_rd - write_loc;
 128
 129
 130         if ((prev_write_sz < pending_sz) && (cur_write_sz >= pending_sz))
 131                 return true;
 132
 133         return false;
 134 }
 135
 136 /*
 137  * hv_get_next_write_location()
 138  *
 139  * Get the next write location for the specified ring buffer
 140  *
 141  */
 142 static inline u32
 143 hv_get_next_write_location(struct hv_ring_buffer_info *ring_info)
 144 {
 145         u32 next = ring_info->ring_buffer->write_index;
 146
 147         return next;
 148 }
 149
 150 /*
 151  * hv_set_next_write_location()
 152  *
 153  * Set the next write location for the specified ring buffer
 154  *
 155  */
 156 static inline void
 157 hv_set_next_write_location(struct hv_ring_buffer_info *ring_info,
 158                      u32 next_write_location)
 159 {
 160         ring_info->ring_buffer->write_index = next_write_location;
 161 }
 162
 163 /*
 164  * hv_get_next_read_location()
 165  *
 166  * Get the next read location for the specified ring buffer
 167  */
 168 static inline u32
 169 hv_get_next_read_location(struct hv_ring_buffer_info *ring_info)
 170 {
 171         u32 next = ring_info->ring_buffer->read_index;
 172
 173         return next;
 174 }
 175
 176 /*
 177  * hv_get_next_readlocation_withoffset()
 178  *
 179  * Get the next read location + offset for the specified ring buffer.
 180  * This allows the caller to skip
 181  */
 182 static inline u32
 183 hv_get_next_readlocation_withoffset(struct hv_ring_buffer_info *ring_info,
 184                                  u32 offset)
 185 {
 186         u32 next = ring_info->ring_buffer->read_index;
 187
 188         next += offset;
 189         next %= ring_info->ring_datasize;
 190
 191         return next;
 192 }
 193
 194 /*
 195  *
 196  * hv_set_next_read_location()
 197  *
 198  * Set the next read location for the specified ring buffer
 199  *
 200  */
 201 static inline void
 202 hv_set_next_read_location(struct hv_ring_buffer_info *ring_info,
 203                     u32 next_read_location)
 204 {
 205         ring_info->ring_buffer->read_index = next_read_location;
 206 }
 207
 208
 209 /*
 210  *
 211  * hv_get_ring_buffer()
 212  *
 213  * Get the start of the ring buffer
 214  */
 215 static inline void *
 216 hv_get_ring_buffer(struct hv_ring_buffer_info *ring_info)
 217 {
 218         return (void *)ring_info->ring_buffer->buffer;
 219 }
 220
 221
 222 /*
 223  *
 224  * hv_get_ring_buffersize()
 225  *
 226  * Get the size of the ring buffer
 227  */
 228 static inline u32
 229 hv_get_ring_buffersize(struct hv_ring_buffer_info *ring_info)
 230 {
 231         return ring_info->ring_datasize;
 232 }
 233
 234 /*
 235  *
 236  * hv_get_ring_bufferindices()
 237  *
 238  * Get the read and write indices as u64 of the specified ring buffer
 239  *
 240  */
 241 static inline u64
 242 hv_get_ring_bufferindices(struct hv_ring_buffer_info *ring_info)
 243 {
 244         return (u64)ring_info->ring_buffer->write_index << 32;
 245 }
 246
 247 /*
 248  *
 249  * hv_copyfrom_ringbuffer()
 250  *
 251  * Helper routine to copy to source from ring buffer.
 252  * Assume there is enough room. Handles wrap-around in src case only!!
 253  *
 254  */
 255 static u32 hv_copyfrom_ringbuffer(
 256         struct hv_ring_buffer_info      *ring_info,
 257         void                            *dest,
 258         u32                             destlen,
 259         u32                             start_read_offset)
 260 {
 261         void *ring_buffer = hv_get_ring_buffer(ring_info);
 262         u32 ring_buffer_size = hv_get_ring_buffersize(ring_info);
 263
 264         u32 frag_len;
 265
 266         /* wrap-around detected at the src */
 267         if (destlen > ring_buffer_size - start_read_offset) {
 268                 frag_len = ring_buffer_size - start_read_offset;
 269
 270                 memcpy(dest, ring_buffer + start_read_offset, frag_len);
 271                 memcpy(dest + frag_len, ring_buffer, destlen - frag_len);
 272         } else
 273
 274                 memcpy(dest, ring_buffer + start_read_offset, destlen);
 275
 276
 277         start_read_offset += destlen;
 278         start_read_offset %= ring_buffer_size;
 279
 280         return start_read_offset;
 281 }
 282
 283
 284 /*
 285  *
 286  * hv_copyto_ringbuffer()
 287  *
 288  * Helper routine to copy from source to ring buffer.
 289  * Assume there is enough room. Handles wrap-around in dest case only!!
 290  *
 291  */
 292 static u32 hv_copyto_ringbuffer(
 293         struct hv_ring_buffer_info      *ring_info,
 294         u32                             start_write_offset,
 295         void                            *src,
 296         u32                             srclen)
 297 {
 298         void *ring_buffer = hv_get_ring_buffer(ring_info);
 299         u32 ring_buffer_size = hv_get_ring_buffersize(ring_info);
 300         u32 frag_len;
 301
 302         /* wrap-around detected! */
 303         if (srclen > ring_buffer_size - start_write_offset) {
 304                 frag_len = ring_buffer_size - start_write_offset;
 305                 memcpy(ring_buffer + start_write_offset, src, frag_len);
 306                 memcpy(ring_buffer, src + frag_len, srclen - frag_len);
 307         } else
 308                 memcpy(ring_buffer + start_write_offset, src, srclen);
 309
 310         start_write_offset += srclen;
 311         start_write_offset %= ring_buffer_size;
 312
 313         return start_write_offset;
 314 }
 315
 316 /*
 317  *
 318  * hv_ringbuffer_get_debuginfo()
 319  *
 320  * Get various debug metrics for the specified ring buffer
 321  *
 322  */
 323 void hv_ringbuffer_get_debuginfo(struct hv_ring_buffer_info *ring_info,
 324                             struct hv_ring_buffer_debug_info *debug_info)
 325 {
 326         u32 bytes_avail_towrite;
 327         u32 bytes_avail_toread;
 328
 329         if (ring_info->ring_buffer) {
 330                 hv_get_ringbuffer_availbytes(ring_info,
 331                                         &bytes_avail_toread,
 332                                         &bytes_avail_towrite);
 333
 334                 debug_info->bytes_avail_toread = bytes_avail_toread;
 335                 debug_info->bytes_avail_towrite = bytes_avail_towrite;
 336                 debug_info->current_read_index =
 337                         ring_info->ring_buffer->read_index;
 338                 debug_info->current_write_index =
 339                         ring_info->ring_buffer->write_index;
 340                 debug_info->current_interrupt_mask =
 341                         ring_info->ring_buffer->interrupt_mask;
 342         }
 343 }
 344
 345 /*
 346  *
 347  * hv_ringbuffer_init()
 348  *
 349  *Initialize the ring buffer
 350  *
 351  */
 352 int hv_ringbuffer_init(struct hv_ring_buffer_info *ring_info,
 353                    void *buffer, u32 buflen)
 354 {
 355         if (sizeof(struct hv_ring_buffer) != PAGE_SIZE)
 356                 return -EINVAL;
 357
 358         memset(ring_info, 0, sizeof(struct hv_ring_buffer_info));
 359
 360         ring_info->ring_buffer = (struct hv_ring_buffer *)buffer;
 361         ring_info->ring_buffer->read_index =
 362                 ring_info->ring_buffer->write_index = 0;
 363
 364         ring_info->ring_size = buflen;
 365         ring_info->ring_datasize = buflen - sizeof(struct hv_ring_buffer);
 366
 367         spin_lock_init(&ring_info->ring_lock);
 368
 369         return 0;
 370 }
 371
 372 /*
 373  *
 374  * hv_ringbuffer_cleanup()
 375  *
 376  * Cleanup the ring buffer
 377  *
 378  */
 379 void hv_ringbuffer_cleanup(struct hv_ring_buffer_info *ring_info)
 380 {
 381 }
 382
 383 /*
 384  *
 385  * hv_ringbuffer_write()
 386  *
 387  * Write to the ring buffer
 388  *
 389  */
 390 int hv_ringbuffer_write(struct hv_ring_buffer_info *outring_info,
 391                     struct kvec *kv_list, u32 kv_count, bool *signal)
 392 {
 393         int i = 0;
 394         u32 bytes_avail_towrite;
 395         u32 bytes_avail_toread;
 396         u32 totalbytes_towrite = 0;
 397
 398         u32 next_write_location;
 399         u32 old_write;
 400         u64 prev_indices = 0;
 401         unsigned long flags;
 402
 403         for (i = 0; i < kv_count; i++)
 404                 totalbytes_towrite += kv_list[i].iov_len;
 405
 406         totalbytes_towrite += sizeof(u64);
 407
 408         spin_lock_irqsave(&outring_info->ring_lock, flags);
 409
 410         hv_get_ringbuffer_availbytes(outring_info,
 411                                 &bytes_avail_toread,
 412                                 &bytes_avail_towrite);
 413
 414
 415         /* If there is only room for the packet, assume it is full. */
 416         /* Otherwise, the next time around, we think the ring buffer */
 417         /* is empty since the read index == write index */
 418         if (bytes_avail_towrite <= totalbytes_towrite) {
 419                 spin_unlock_irqrestore(&outring_info->ring_lock, flags);
 420                 return -EAGAIN;
 421         }
 422
 423         /* Write to the ring buffer */
 424         next_write_location = hv_get_next_write_location(outring_info);
 425
 426         old_write = next_write_location;
 427
 428         for (i = 0; i < kv_count; i++) {
 429                 next_write_location = hv_copyto_ringbuffer(outring_info,
 430                                                      next_write_location,
 431                                                      kv_list[i].iov_base,
 432                                                      kv_list[i].iov_len);
 433         }
 434
 435         /* Set previous packet start */
 436         prev_indices = hv_get_ring_bufferindices(outring_info);
 437
 438         next_write_location = hv_copyto_ringbuffer(outring_info,
 439                                              next_write_location,
 440                                              &prev_indices,
 441                                              sizeof(u64));
 442
 443         /* Issue a full memory barrier before updating the write index */
 444         mb();
 445
 446         /* Now, update the write location */
 447         hv_set_next_write_location(outring_info, next_write_location);
 448
 449
 450         spin_unlock_irqrestore(&outring_info->ring_lock, flags);
 451
 452         *signal = hv_need_to_signal(old_write, outring_info);
 453         return 0;
 454 }
 455
 456
 457 /*
 458  *
 459  * hv_ringbuffer_peek()
 460  *
 461  * Read without advancing the read index
 462  *
 463  */
 464 int hv_ringbuffer_peek(struct hv_ring_buffer_info *Inring_info,
 465                    void *Buffer, u32 buflen)
 466 {
 467         u32 bytes_avail_towrite;
 468         u32 bytes_avail_toread;
 469         u32 next_read_location = 0;
 470         unsigned long flags;
 471
 472         spin_lock_irqsave(&Inring_info->ring_lock, flags);
 473
 474         hv_get_ringbuffer_availbytes(Inring_info,
 475                                 &bytes_avail_toread,
 476                                 &bytes_avail_towrite);
 477
 478         /* Make sure there is something to read */
 479         if (bytes_avail_toread < buflen) {
 480
 481                 spin_unlock_irqrestore(&Inring_info->ring_lock, flags);
 482
 483                 return -EAGAIN;
 484         }
 485
 486         /* Convert to byte offset */
 487         next_read_location = hv_get_next_read_location(Inring_info);
 488
 489         next_read_location = hv_copyfrom_ringbuffer(Inring_info,
 490                                                 Buffer,
 491                                                 buflen,
 492                                                 next_read_location);
 493
 494         spin_unlock_irqrestore(&Inring_info->ring_lock, flags);
 495
 496         return 0;
 497 }
 498
 499
 500 /*
 501  *
 502  * hv_ringbuffer_read()
 503  *
 504  * Read and advance the read index
 505  *
 506  */
 507 int hv_ringbuffer_read(struct hv_ring_buffer_info *inring_info, void *buffer,
 508                    u32 buflen, u32 offset, bool *signal)
 509 {
 510         u32 bytes_avail_towrite;
 511         u32 bytes_avail_toread;
 512         u32 next_read_location = 0;
 513         u64 prev_indices = 0;
 514         unsigned long flags;
 515         u32 old_read;
 516
 517         if (buflen <= 0)
 518                 return -EINVAL;
 519
 520         spin_lock_irqsave(&inring_info->ring_lock, flags);
 521
 522         hv_get_ringbuffer_availbytes(inring_info,
 523                                 &bytes_avail_toread,
 524                                 &bytes_avail_towrite);
 525
 526         old_read = bytes_avail_toread;
 527
 528         /* Make sure there is something to read */
 529         if (bytes_avail_toread < buflen) {
 530                 spin_unlock_irqrestore(&inring_info->ring_lock, flags);
 531
 532                 return -EAGAIN;
 533         }
 534
 535         next_read_location =
 536                 hv_get_next_readlocation_withoffset(inring_info, offset);
 537
 538         next_read_location = hv_copyfrom_ringbuffer(inring_info,
 539                                                 buffer,
 540                                                 buflen,
 541                                                 next_read_location);
 542
 543         next_read_location = hv_copyfrom_ringbuffer(inring_info,
 544                                                 &prev_indices,
 545                                                 sizeof(u64),
 546                                                 next_read_location);
 547
 548         /* Make sure all reads are done before we update the read index since */
 549         /* the writer may start writing to the read area once the read index */
 550         /*is updated */
 551         mb();
 552
 553         /* Update the read index */
 554         hv_set_next_read_location(inring_info, next_read_location);
 555
 556         spin_unlock_irqrestore(&inring_info->ring_lock, flags);
 557
 558         *signal = hv_need_to_signal_on_read(old_read, inring_info);
 559
 560         return 0;
 561 }