ib_ring.c 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168
  1. /*
  2. * Copyright (c) 2006 Oracle. All rights reserved.
  3. *
  4. * This software is available to you under a choice of one of two
  5. * licenses. You may choose to be licensed under the terms of the GNU
  6. * General Public License (GPL) Version 2, available from the file
  7. * COPYING in the main directory of this source tree, or the
  8. * OpenIB.org BSD license below:
  9. *
  10. * Redistribution and use in source and binary forms, with or
  11. * without modification, are permitted provided that the following
  12. * conditions are met:
  13. *
  14. * - Redistributions of source code must retain the above
  15. * copyright notice, this list of conditions and the following
  16. * disclaimer.
  17. *
  18. * - Redistributions in binary form must reproduce the above
  19. * copyright notice, this list of conditions and the following
  20. * disclaimer in the documentation and/or other materials
  21. * provided with the distribution.
  22. *
  23. * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
  24. * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  25. * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
  26. * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
  27. * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
  28. * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
  29. * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  30. * SOFTWARE.
  31. *
  32. */
  33. #include <linux/kernel.h>
  34. #include "rds.h"
  35. #include "ib.h"
  36. /*
  37. * Locking for IB rings.
  38. * We assume that allocation is always protected by a mutex
  39. * in the caller (this is a valid assumption for the current
  40. * implementation).
  41. *
  42. * Freeing always happens in an interrupt, and hence only
  43. * races with allocations, but not with other free()s.
  44. *
  45. * The interaction between allocation and freeing is that
  46. * the alloc code has to determine the number of free entries.
  47. * To this end, we maintain two counters; an allocation counter
  48. * and a free counter. Both are allowed to run freely, and wrap
  49. * around.
  50. * The number of used entries is always (alloc_ctr - free_ctr) % NR.
  51. *
  52. * The current implementation makes free_ctr atomic. When the
  53. * caller finds an allocation fails, it should set an "alloc fail"
  54. * bit and retry the allocation. The "alloc fail" bit essentially tells
  55. * the CQ completion handlers to wake it up after freeing some
  56. * more entries.
  57. */
  58. /*
  59. * This only happens on shutdown.
  60. */
  61. DECLARE_WAIT_QUEUE_HEAD(rds_ib_ring_empty_wait);
  62. void rds_ib_ring_init(struct rds_ib_work_ring *ring, u32 nr)
  63. {
  64. memset(ring, 0, sizeof(*ring));
  65. ring->w_nr = nr;
  66. rdsdebug("ring %p nr %u\n", ring, ring->w_nr);
  67. }
  68. static inline u32 __rds_ib_ring_used(struct rds_ib_work_ring *ring)
  69. {
  70. u32 diff;
  71. /* This assumes that atomic_t has at least as many bits as u32 */
  72. diff = ring->w_alloc_ctr - (u32) atomic_read(&ring->w_free_ctr);
  73. BUG_ON(diff > ring->w_nr);
  74. return diff;
  75. }
  76. void rds_ib_ring_resize(struct rds_ib_work_ring *ring, u32 nr)
  77. {
  78. /* We only ever get called from the connection setup code,
  79. * prior to creating the QP. */
  80. BUG_ON(__rds_ib_ring_used(ring));
  81. ring->w_nr = nr;
  82. }
  83. static int __rds_ib_ring_empty(struct rds_ib_work_ring *ring)
  84. {
  85. return __rds_ib_ring_used(ring) == 0;
  86. }
  87. u32 rds_ib_ring_alloc(struct rds_ib_work_ring *ring, u32 val, u32 *pos)
  88. {
  89. u32 ret = 0, avail;
  90. avail = ring->w_nr - __rds_ib_ring_used(ring);
  91. rdsdebug("ring %p val %u next %u free %u\n", ring, val,
  92. ring->w_alloc_ptr, avail);
  93. if (val && avail) {
  94. ret = min(val, avail);
  95. *pos = ring->w_alloc_ptr;
  96. ring->w_alloc_ptr = (ring->w_alloc_ptr + ret) % ring->w_nr;
  97. ring->w_alloc_ctr += ret;
  98. }
  99. return ret;
  100. }
  101. void rds_ib_ring_free(struct rds_ib_work_ring *ring, u32 val)
  102. {
  103. ring->w_free_ptr = (ring->w_free_ptr + val) % ring->w_nr;
  104. atomic_add(val, &ring->w_free_ctr);
  105. if (__rds_ib_ring_empty(ring) &&
  106. waitqueue_active(&rds_ib_ring_empty_wait))
  107. wake_up(&rds_ib_ring_empty_wait);
  108. }
  109. void rds_ib_ring_unalloc(struct rds_ib_work_ring *ring, u32 val)
  110. {
  111. ring->w_alloc_ptr = (ring->w_alloc_ptr - val) % ring->w_nr;
  112. ring->w_alloc_ctr -= val;
  113. }
  114. int rds_ib_ring_empty(struct rds_ib_work_ring *ring)
  115. {
  116. return __rds_ib_ring_empty(ring);
  117. }
  118. int rds_ib_ring_low(struct rds_ib_work_ring *ring)
  119. {
  120. return __rds_ib_ring_used(ring) <= (ring->w_nr >> 1);
  121. }
  122. /*
  123. * returns the oldest alloced ring entry. This will be the next one
  124. * freed. This can't be called if there are none allocated.
  125. */
  126. u32 rds_ib_ring_oldest(struct rds_ib_work_ring *ring)
  127. {
  128. return ring->w_free_ptr;
  129. }
  130. /*
  131. * returns the number of completed work requests.
  132. */
  133. u32 rds_ib_ring_completed(struct rds_ib_work_ring *ring, u32 wr_id, u32 oldest)
  134. {
  135. u32 ret;
  136. if (oldest <= (unsigned long long)wr_id)
  137. ret = (unsigned long long)wr_id - oldest + 1;
  138. else
  139. ret = ring->w_nr - oldest + (unsigned long long)wr_id + 1;
  140. rdsdebug("ring %p ret %u wr_id %u oldest %u\n", ring, ret,
  141. wr_id, oldest);
  142. return ret;
  143. }