MCPcopy Create free account
hub / github.com/F-Stack/f-stack / handle_worker_shutdown

Function handle_worker_shutdown

dpdk/lib/distributor/rte_distributor.c:287–332  ·  view source on GitHub ↗

* When worker called rte_distributor_return_pkt() * and passed RTE_DISTRIB_RETURN_BUF handshake through retptr64, * distributor must retrieve both inflight and backlog packets assigned * to the worker and reprocess them to another worker. */

Source from the content-addressed store, hash-verified

285 * to the worker and reprocess them to another worker.
286 */
287static void
288handle_worker_shutdown(struct rte_distributor *d, unsigned int wkr)
289{
290 struct rte_distributor_buffer *buf = &(d->bufs[wkr]);
291 /* double BURST size for storing both inflights and backlog */
292 struct rte_mbuf *pkts[RTE_DIST_BURST_SIZE * 2];
293 unsigned int pkts_count = 0;
294 unsigned int i;
295
296 /* If GET_BUF is cleared there are in-flight packets sent
297 * to worker which does not require new packets.
298 * They must be retrieved and assigned to another worker.
299 */
300 if (!(rte_atomic_load_explicit(&(buf->bufptr64[0]), rte_memory_order_acquire)
301 & RTE_DISTRIB_GET_BUF))
302 for (i = 0; i < RTE_DIST_BURST_SIZE; i++)
303 if (buf->bufptr64[i] & RTE_DISTRIB_VALID_BUF)
304 pkts[pkts_count++] = (void *)((uintptr_t)
305 (buf->bufptr64[i]
306 >> RTE_DISTRIB_FLAG_BITS));
307
308 /* Make following operations on handshake flags on bufptr64:
309 * - set GET_BUF to indicate that distributor can overwrite buffer
310 * with new packets if worker will make a new request.
311 * - clear RETURN_BUF to unlock reads on worker side.
312 */
313 rte_atomic_store_explicit(&(buf->bufptr64[0]), RTE_DISTRIB_GET_BUF,
314 rte_memory_order_release);
315
316 /* Collect backlog packets from worker */
317 for (i = 0; i < d->backlog[wkr].count; i++)
318 pkts[pkts_count++] = (void *)((uintptr_t)
319 (d->backlog[wkr].pkts[i] >> RTE_DISTRIB_FLAG_BITS));
320
321 d->backlog[wkr].count = 0;
322
323 /* Clear both inflight and backlog tags */
324 for (i = 0; i < RTE_DIST_BURST_SIZE; i++) {
325 d->in_flight_tags[wkr][i] = 0;
326 d->backlog[wkr].tags[i] = 0;
327 }
328
329 /* Recursive call */
330 if (pkts_count > 0)
331 rte_distributor_process(d, pkts, pkts_count);
332}
333
334
335/*

Callers 1

handle_returnsFunction · 0.70

Calls 1

rte_distributor_processFunction · 0.85

Tested by

no test coverage detected