X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=src%2Fplugins%2Fmemif%2Fdevice.c;h=5c8eb3f082c6a5834798d71daeee2053ccec953c;hb=067cd6229a47ea3ba8b59a2a04090e80afb5bd2c;hp=5f670563fcbca049df3dbc8fffab4d729b2d6025;hpb=7172cec2384e107a8127a2cb8cc346ef9216d425;p=vpp.git diff --git a/src/plugins/memif/device.c b/src/plugins/memif/device.c index 5f670563fcb..5c8eb3f082c 100644 --- a/src/plugins/memif/device.c +++ b/src/plugins/memif/device.c @@ -30,8 +30,7 @@ #define foreach_memif_tx_func_error \ _(NO_FREE_SLOTS, "no free tx slots") \ -_(TRUNC_PACKET, "packet > buffer size -- truncated in tx ring") \ -_(PENDING_MSGS, "pending msgs in tx ring") +_(ROLLBACK, "no enough space in tx buffers") typedef enum { @@ -47,15 +46,17 @@ static __clib_unused char *memif_tx_func_error_strings[] = { #undef _ }; -#ifndef CLIB_MULTIARCH_VARIANT +#ifndef CLIB_MARCH_VARIANT u8 * format_memif_device_name (u8 * s, va_list * args) { u32 dev_instance = va_arg (*args, u32); memif_main_t *mm = &memif_main; memif_if_t *mif = pool_elt_at_index (mm->interfaces, dev_instance); + memif_socket_file_t *msf; - s = format (s, "memif%lu/%lu", mif->socket_file_index, mif->id); + msf = pool_elt_at_index (mm->socket_files, mif->socket_file_index); + s = format (s, "memif%lu/%lu", msf->socket_id, mif->id); return s; } #endif @@ -84,151 +85,298 @@ format_memif_tx_trace (u8 * s, va_list * args) } static_always_inline void -memif_prefetch_buffer_and_data (vlib_main_t * vm, u32 bi) +memif_add_copy_op (memif_per_thread_data_t * ptd, void *data, u32 len, + u16 buffer_offset, u16 buffer_vec_index) { - vlib_buffer_t *b = vlib_get_buffer (vm, bi); - vlib_prefetch_buffer_header (b, LOAD); - CLIB_PREFETCH (b->data, CLIB_CACHE_LINE_BYTES, LOAD); + memif_copy_op_t *co; + vec_add2_aligned (ptd->copy_ops, co, 1, CLIB_CACHE_LINE_BYTES); + co->data = data; + co->data_len = len; + co->buffer_offset = buffer_offset; + co->buffer_vec_index = buffer_vec_index; } -/** - * @brief Copy buffer to tx ring - * - * @param * vm (in) - * @param * node (in) - * @param * mif (in) pointer to memif interface - * @param bi (in) vlib buffer index - * @param * ring (in) pointer to memif ring - * @param * head (in/out) ring head - * @param mask (in) ring size - 1 - */ -static_always_inline void -memif_copy_buffer_to_tx_ring (vlib_main_t * vm, vlib_node_runtime_t * node, - memif_if_t * mif, u32 bi, memif_ring_t * ring, - u16 * head, u16 mask) +static_always_inline uword +memif_interface_tx_inline (vlib_main_t * vm, vlib_node_runtime_t * node, + vlib_frame_t * frame, memif_if_t * mif, + memif_ring_type_t type, memif_queue_t * mq, + memif_per_thread_data_t * ptd) { - vlib_buffer_t *b0; - void *mb0; - u32 total = 0, len; - u16 slot = (*head) & mask; + memif_ring_t *ring; + u32 *buffers = vlib_frame_args (frame); + u32 n_left = frame->n_vectors; + u32 n_copy_op; + u16 ring_size, mask, slot, free_slots; + int n_retries = 5; + vlib_buffer_t *b0, *b1, *b2, *b3; + memif_copy_op_t *co; + memif_region_index_t last_region = ~0; + void *last_region_shm = 0; + + ring = mq->ring; + ring_size = 1 << mq->log2_ring_size; + mask = ring_size - 1; + +retry: + + free_slots = ring->tail - mq->last_tail; + mq->last_tail += free_slots; + slot = (type == MEMIF_RING_S2M) ? ring->head : ring->tail; - mb0 = memif_get_buffer (mif, ring, slot); - ring->desc[slot].flags = 0; - do + if (type == MEMIF_RING_S2M) + free_slots = ring_size - ring->head + mq->last_tail; + else + free_slots = ring->head - ring->tail; + + while (n_left && free_slots) { - b0 = vlib_get_buffer (vm, bi); - len = b0->current_length; - if (PREDICT_FALSE (ring->desc[slot].buffer_length < (total + len))) + memif_desc_t *d0; + void *mb0; + i32 src_off; + u32 bi0, dst_off, src_left, dst_left, bytes_to_copy; + u32 saved_ptd_copy_ops_len = _vec_len (ptd->copy_ops); + u32 saved_ptd_buffers_len = _vec_len (ptd->buffers); + u16 saved_slot = slot; + + CLIB_PREFETCH (&ring->desc[(slot + 8) & mask], CLIB_CACHE_LINE_BYTES, + LOAD); + + d0 = &ring->desc[slot & mask]; + if (PREDICT_FALSE (last_region != d0->region)) { - if (PREDICT_TRUE (total)) - { - ring->desc[slot].length = total; - total = 0; - ring->desc[slot].flags |= MEMIF_DESC_FLAG_NEXT; - (*head)++; - slot = (*head) & mask; - mb0 = memif_get_buffer (mif, ring, slot); - ring->desc[slot].flags = 0; - } + last_region_shm = mif->regions[d0->region].shm; + last_region = d0->region; } - if (PREDICT_TRUE (ring->desc[slot].buffer_length >= (total + len))) + mb0 = last_region_shm + d0->offset; + + dst_off = 0; + + /* slave is the producer, so it should be able to reset buffer length */ + dst_left = (type == MEMIF_RING_S2M) ? mif->run.buffer_size : d0->length; + + if (PREDICT_TRUE (n_left >= 4)) + vlib_prefetch_buffer_header (vlib_get_buffer (vm, buffers[3]), LOAD); + bi0 = buffers[0]; + + next_in_chain: + + b0 = vlib_get_buffer (vm, bi0); + src_off = b0->current_data; + src_left = b0->current_length; + + while (src_left) { - clib_memcpy (mb0 + total, vlib_buffer_get_current (b0), - CLIB_CACHE_LINE_BYTES); - if (len > CLIB_CACHE_LINE_BYTES) - clib_memcpy (mb0 + CLIB_CACHE_LINE_BYTES + total, - vlib_buffer_get_current (b0) + CLIB_CACHE_LINE_BYTES, - len - CLIB_CACHE_LINE_BYTES); - total += len; + if (PREDICT_FALSE (dst_left == 0)) + { + if (free_slots) + { + slot++; + free_slots--; + d0->flags = MEMIF_DESC_FLAG_NEXT; + d0 = &ring->desc[slot & mask]; + dst_off = 0; + dst_left = + (type == + MEMIF_RING_S2M) ? mif->run.buffer_size : d0->length; + + if (PREDICT_FALSE (last_region != d0->region)) + { + last_region_shm = mif->regions[d0->region].shm; + last_region = d0->region; + } + mb0 = last_region_shm + d0->offset; + } + else + { + /* we need to rollback vectors before bailing out */ + _vec_len (ptd->buffers) = saved_ptd_buffers_len; + _vec_len (ptd->copy_ops) = saved_ptd_copy_ops_len; + vlib_error_count (vm, node->node_index, + MEMIF_TX_ERROR_ROLLBACK, 1); + slot = saved_slot; + goto no_free_slots; + } + } + bytes_to_copy = clib_min (src_left, dst_left); + memif_add_copy_op (ptd, mb0 + dst_off, bytes_to_copy, src_off, + vec_len (ptd->buffers)); + vec_add1_aligned (ptd->buffers, bi0, CLIB_CACHE_LINE_BYTES); + src_off += bytes_to_copy; + dst_off += bytes_to_copy; + src_left -= bytes_to_copy; + dst_left -= bytes_to_copy; } - else + + if (PREDICT_FALSE (b0->flags & VLIB_BUFFER_NEXT_PRESENT)) { - vlib_error_count (vm, node->node_index, MEMIF_TX_ERROR_TRUNC_PACKET, - 1); - break; + bi0 = b0->next_buffer; + goto next_in_chain; } + + d0->length = dst_off; + d0->flags = 0; + + free_slots -= 1; + slot += 1; + + buffers++; + n_left--; } - while ((bi = (b0->flags & VLIB_BUFFER_NEXT_PRESENT) ? b0->next_buffer : 0)); +no_free_slots: - if (PREDICT_TRUE (total)) + /* copy data */ + n_copy_op = vec_len (ptd->copy_ops); + co = ptd->copy_ops; + while (n_copy_op >= 8) + { + CLIB_PREFETCH (co[4].data, CLIB_CACHE_LINE_BYTES, LOAD); + CLIB_PREFETCH (co[5].data, CLIB_CACHE_LINE_BYTES, LOAD); + CLIB_PREFETCH (co[6].data, CLIB_CACHE_LINE_BYTES, LOAD); + CLIB_PREFETCH (co[7].data, CLIB_CACHE_LINE_BYTES, LOAD); + + b0 = vlib_get_buffer (vm, ptd->buffers[co[0].buffer_vec_index]); + b1 = vlib_get_buffer (vm, ptd->buffers[co[1].buffer_vec_index]); + b2 = vlib_get_buffer (vm, ptd->buffers[co[2].buffer_vec_index]); + b3 = vlib_get_buffer (vm, ptd->buffers[co[3].buffer_vec_index]); + + clib_memcpy (co[0].data, b0->data + co[0].buffer_offset, + co[0].data_len); + clib_memcpy (co[1].data, b1->data + co[1].buffer_offset, + co[1].data_len); + clib_memcpy (co[2].data, b2->data + co[2].buffer_offset, + co[2].data_len); + clib_memcpy (co[3].data, b3->data + co[3].buffer_offset, + co[3].data_len); + + co += 4; + n_copy_op -= 4; + } + while (n_copy_op) { - ring->desc[slot].length = total; - (*head)++; + b0 = vlib_get_buffer (vm, ptd->buffers[co[0].buffer_vec_index]); + clib_memcpy (co[0].data, b0->data + co[0].buffer_offset, + co[0].data_len); + co += 1; + n_copy_op -= 1; } + + vec_reset_length (ptd->copy_ops); + vec_reset_length (ptd->buffers); + + CLIB_MEMORY_STORE_BARRIER (); + if (type == MEMIF_RING_S2M) + ring->head = slot; + else + ring->tail = slot; + + if (n_left && n_retries--) + goto retry; + + clib_spinlock_unlock_if_init (&mif->lockp); + + if (n_left) + { + vlib_error_count (vm, node->node_index, MEMIF_TX_ERROR_NO_FREE_SLOTS, + n_left); + } + + if ((ring->flags & MEMIF_RING_FLAG_MASK_INT) == 0 && mq->int_fd > -1) + { + u64 b = 1; + CLIB_UNUSED (int r) = write (mq->int_fd, &b, sizeof (b)); + mq->int_count++; + } + + vlib_buffer_free (vm, vlib_frame_args (frame), frame->n_vectors); + + return frame->n_vectors; } static_always_inline uword -memif_interface_tx_inline (vlib_main_t * vm, vlib_node_runtime_t * node, - vlib_frame_t * frame, memif_if_t * mif, - memif_ring_type_t type) +memif_interface_tx_zc_inline (vlib_main_t * vm, vlib_node_runtime_t * node, + vlib_frame_t * frame, memif_if_t * mif, + memif_queue_t * mq, + memif_per_thread_data_t * ptd) { - u8 qid; - memif_ring_t *ring; + memif_ring_t *ring = mq->ring; u32 *buffers = vlib_frame_args (frame); u32 n_left = frame->n_vectors; - u16 ring_size, mask; - u16 head, tail; - u16 free_slots; - u32 thread_index = vlib_get_thread_index (); - u8 tx_queues = vec_len (mif->tx_queues); - memif_queue_t *mq; + u16 slot, free_slots, n_free; + u16 ring_size = 1 << mq->log2_ring_size; + u16 mask = ring_size - 1; int n_retries = 5; + vlib_buffer_t *b0; - if (tx_queues < vec_len (vlib_mains)) +retry: + n_free = ring->tail - mq->last_tail; + if (n_free >= 16) { - qid = thread_index % tx_queues; - clib_spinlock_lock_if_init (&mif->lockp); + vlib_buffer_free_from_ring_no_next (vm, mq->buffers, + mq->last_tail & mask, + ring_size, n_free); + mq->last_tail += n_free; } - else - qid = thread_index; - mq = vec_elt_at_index (mif->tx_queues, qid); - ring = mq->ring; - ring_size = 1 << mq->log2_ring_size; - mask = ring_size - 1; -retry: + slot = ring->head; + free_slots = ring_size - ring->head + mq->last_tail; + + while (n_left && free_slots) + { + u16 s0; + u16 slots_in_packet = 1; + memif_desc_t *d0; + u32 bi0; - /* free consumed buffers */ + CLIB_PREFETCH (&ring->desc[(slot + 8) & mask], CLIB_CACHE_LINE_BYTES, + STORE); - head = ring->head; - tail = ring->tail; + if (PREDICT_TRUE (n_left >= 4)) + vlib_prefetch_buffer_header (vlib_get_buffer (vm, buffers[3]), LOAD); - free_slots = ring_size - head + tail; + bi0 = buffers[0]; - while (n_left > 5 && free_slots > 1) - { - CLIB_PREFETCH (memif_get_buffer (mif, ring, (head + 2) & mask), - CLIB_CACHE_LINE_BYTES, STORE); - CLIB_PREFETCH (memif_get_buffer (mif, ring, (head + 3) & mask), - CLIB_CACHE_LINE_BYTES, STORE); - CLIB_PREFETCH (&ring->desc[(head + 4) & mask], CLIB_CACHE_LINE_BYTES, - STORE); - CLIB_PREFETCH (&ring->desc[(head + 5) & mask], CLIB_CACHE_LINE_BYTES, - STORE); - memif_prefetch_buffer_and_data (vm, buffers[2]); - memif_prefetch_buffer_and_data (vm, buffers[3]); + next_in_chain: + s0 = slot & mask; + d0 = &ring->desc[s0]; + mq->buffers[s0] = bi0; + b0 = vlib_get_buffer (vm, bi0); - memif_copy_buffer_to_tx_ring (vm, node, mif, buffers[0], ring, &head, - mask); - memif_copy_buffer_to_tx_ring (vm, node, mif, buffers[1], ring, &head, - mask); + d0->region = b0->buffer_pool_index + 1; + d0->offset = (void *) b0->data + b0->current_data - + mif->regions[d0->region].shm; + d0->length = b0->current_length; - buffers += 2; - n_left -= 2; - free_slots -= 2; - } + free_slots--; + slot++; - while (n_left && free_slots) - { - memif_copy_buffer_to_tx_ring (vm, node, mif, buffers[0], ring, &head, - mask); + if (PREDICT_FALSE (b0->flags & VLIB_BUFFER_NEXT_PRESENT)) + { + if (PREDICT_FALSE (free_slots == 0)) + { + /* revert to last fully processed packet */ + free_slots += slots_in_packet; + slot -= slots_in_packet; + goto no_free_slots; + } + + d0->flags = MEMIF_DESC_FLAG_NEXT; + bi0 = b0->next_buffer; + + /* next */ + slots_in_packet++; + goto next_in_chain; + } + + d0->flags = 0; + + /* next from */ buffers++; n_left--; - free_slots--; } +no_free_slots: CLIB_MEMORY_STORE_BARRIER (); - ring->head = head; + ring->head = slot; if (n_left && n_retries--) goto retry; @@ -239,6 +387,7 @@ retry: { vlib_error_count (vm, node->node_index, MEMIF_TX_ERROR_NO_FREE_SLOTS, n_left); + vlib_buffer_free (vm, buffers, n_left); } if ((ring->flags & MEMIF_RING_FLAG_MASK_INT) == 0 && mq->int_fd > -1) @@ -248,8 +397,6 @@ retry: mq->int_count++; } - vlib_buffer_free (vm, vlib_frame_args (frame), frame->n_vectors); - return frame->n_vectors; } @@ -261,11 +408,29 @@ CLIB_MULTIARCH_FN (memif_interface_tx) (vlib_main_t * vm, memif_main_t *nm = &memif_main; vnet_interface_output_runtime_t *rund = (void *) node->runtime_data; memif_if_t *mif = pool_elt_at_index (nm->interfaces, rund->dev_instance); + memif_queue_t *mq; + u32 thread_index = vm->thread_index; + memif_per_thread_data_t *ptd = vec_elt_at_index (memif_main.per_thread_data, + thread_index); + u8 tx_queues = vec_len (mif->tx_queues); + + if (tx_queues < vec_len (vlib_mains)) + { + ASSERT (tx_queues > 0); + mq = vec_elt_at_index (mif->tx_queues, thread_index % tx_queues); + clib_spinlock_lock_if_init (&mif->lockp); + } + else + mq = vec_elt_at_index (mif->tx_queues, thread_index); - if (mif->flags & MEMIF_IF_FLAG_IS_SLAVE) - return memif_interface_tx_inline (vm, node, frame, mif, MEMIF_RING_S2M); + if (mif->flags & MEMIF_IF_FLAG_ZERO_COPY) + return memif_interface_tx_zc_inline (vm, node, frame, mif, mq, ptd); + else if (mif->flags & MEMIF_IF_FLAG_IS_SLAVE) + return memif_interface_tx_inline (vm, node, frame, mif, MEMIF_RING_S2M, + mq, ptd); else - return memif_interface_tx_inline (vm, node, frame, mif, MEMIF_RING_M2S); + return memif_interface_tx_inline (vm, node, frame, mif, MEMIF_RING_M2S, + mq, ptd); } static __clib_unused void @@ -335,7 +500,7 @@ memif_subif_add_del_function (vnet_main_t * vnm, return 0; } -#ifndef CLIB_MULTIARCH_VARIANT +#ifndef CLIB_MARCH_VARIANT /* *INDENT-OFF* */ VNET_DEVICE_CLASS (memif_device_class) = { .name = "memif", @@ -356,7 +521,7 @@ VNET_DEVICE_CLASS (memif_device_class) = { vlib_node_function_t __clib_weak memif_interface_tx_avx512; vlib_node_function_t __clib_weak memif_interface_tx_avx2; static void __clib_constructor -dpdk_interface_tx_multiarch_select (void) +memif_interface_tx_multiarch_select (void) { if (memif_interface_tx_avx512 && clib_cpu_supports_avx512f ()) memif_device_class.tx_function = memif_interface_tx_avx512;