[dpdk-dev] [PATCH 1/6] service cores: header and implementation

[Jerin Jacob]

-----Original Message-----
> Date: Fri, 23 Jun 2017 10:06:14 +0100
> From: Harry van Haaren <harry.van.haaren at intel.com>
> To: dev at dpdk.org
> CC: thomas at monjalon.net, jerin.jacob at caviumnetworks.com,
>  keith.wiles at intel.com, bruce.richardson at intel.com, Harry van Haaren
>  <harry.van.haaren at intel.com>
> Subject: [PATCH 1/6] service cores: header and implementation
> X-Mailer: git-send-email 2.7.4
> 
> Add header files, update .map files with new service
> functions, and add the service header to the doxygen
> for building.
> 
> This service header API allows DPDK to use services as
> a concept of something that requires CPU cycles. An example
> is a PMD that runs in software to schedule events, where a
> hardware version exists that does not require a CPU.
> 
> The code presented here is based on an initial RFC:
> http://dpdk.org/ml/archives/dev/2017-May/065207.html
> 
> This was then reworked, and RFC v2 with the changes posted:
> http://dpdk.org/ml/archives/dev/2017-June/067194.html
> 
> This is the third iteration of the service core concept,
> now with an implementation.
> 
> Signed-off-by: Harry van Haaren <harry.van.haaren at intel.com>

Nice work. Detailed review comments below

> ---
>  doc/api/doxy-api-index.md                          |   1 +
>  lib/librte_eal/bsdapp/eal/Makefile                 |   1 +
>  lib/librte_eal/bsdapp/eal/rte_eal_version.map      |  24 +
>  lib/librte_eal/common/Makefile                     |   1 +
>  lib/librte_eal/common/include/rte_eal.h            |   4 +
>  lib/librte_eal/common/include/rte_lcore.h          |   3 +-
>  lib/librte_eal/common/include/rte_service.h        | 274 ++++++++++
>  .../common/include/rte_service_private.h           | 108 ++++
>  lib/librte_eal/common/rte_service.c                | 568 +++++++++++++++++++++
>  lib/librte_eal/linuxapp/eal/Makefile               |   1 +
>  lib/librte_eal/linuxapp/eal/rte_eal_version.map    |  24 +
>  11 files changed, 1008 insertions(+), 1 deletion(-)
>  create mode 100644 lib/librte_eal/common/include/rte_service.h
>  create mode 100644 lib/librte_eal/common/include/rte_service_private.h
>  create mode 100644 lib/librte_eal/common/rte_service.c
> 
> diff --git a/doc/api/doxy-api-index.md b/doc/api/doxy-api-index.md
> index f5f1f19..55d522a 100644
> --- a/doc/api/doxy-api-index.md
> +++ b/doc/api/doxy-api-index.md
> @@ -158,6 +158,7 @@ There are many libraries, so their headers may be grouped by topics:
>    [common]             (@ref rte_common.h),
>    [ABI compat]         (@ref rte_compat.h),
>    [keepalive]          (@ref rte_keepalive.h),
> +  [Service Cores]      (@ref rte_service.h),

1) IMO, To keep the consistency we can rename to "[service cores]"
2) I thought, we decided to expose rte_service_register() and
rte_service_unregister() as well, Considering the case where even application
as register for service functions if required. If it is true then I
think, registration functions can moved of private header file so that
it will visible in doxygen.
3) Should we change core function name as lcore like
rte_service_lcore_add(), rte_service_lcore_del() etc as we are operating
on lcore here.


>    [device metrics]     (@ref rte_metrics.h),
>    [bitrate statistics] (@ref rte_bitrate.h),
>    [latency statistics] (@ref rte_latencystats.h),
> diff --git a/lib/librte_eal/bsdapp/eal/Makefile b/lib/librte_eal/bsdapp/eal/Makefile
> index a0f9950..05517a2 100644
> --- a/lib/librte_eal/bsdapp/eal/Makefile
> +++ b/lib/librte_eal/bsdapp/eal/Makefile
> @@ -87,6 +87,7 @@ SRCS-$(CONFIG_RTE_EXEC_ENV_BSDAPP) += rte_malloc.c
>  SRCS-$(CONFIG_RTE_EXEC_ENV_BSDAPP) += malloc_elem.c
>  SRCS-$(CONFIG_RTE_EXEC_ENV_BSDAPP) += malloc_heap.c
>  SRCS-$(CONFIG_RTE_EXEC_ENV_BSDAPP) += rte_keepalive.c
> +SRCS-$(CONFIG_RTE_EXEC_ENV_BSDAPP) += rte_service.c
>  
>  # from arch dir
>  SRCS-$(CONFIG_RTE_EXEC_ENV_BSDAPP) += rte_cpuflags.c
> diff --git a/lib/librte_eal/bsdapp/eal/rte_eal_version.map b/lib/librte_eal/bsdapp/eal/rte_eal_version.map
> index 2e48a73..843d4ee 100644
> --- a/lib/librte_eal/bsdapp/eal/rte_eal_version.map
> +++ b/lib/librte_eal/bsdapp/eal/rte_eal_version.map
> @@ -193,3 +193,27 @@ DPDK_17.05 {
>  	vfio_get_group_no;
>  
>  } DPDK_17.02;
> +
> +DPDK_17.08 {
> +	global:
> +
> +	rte_service_core_add;
> +	rte_service_core_count;
> +	rte_service_core_del;
> +	rte_service_core_list;
> +	rte_service_core_reset_all;
> +	rte_service_core_start;
> +	rte_service_core_stop;
> +	rte_service_disable_on_core;
> +	rte_service_enable_on_core;
> +	rte_service_get_by_id;
> +	rte_service_get_count;
> +	rte_service_get_enabled_on_core;
> +	rte_service_is_running;
> +	rte_service_register;
> +	rte_service_reset;
> +	rte_service_start;
> +	rte_service_stop;
> +	rte_service_unregister;
> +
> +} DPDK_17.05;
> diff --git a/lib/librte_eal/common/Makefile b/lib/librte_eal/common/Makefile
> index a5bd108..2a93397 100644
> --- a/lib/librte_eal/common/Makefile
> +++ b/lib/librte_eal/common/Makefile
> @@ -41,6 +41,7 @@ INC += rte_eal_memconfig.h rte_malloc_heap.h
>  INC += rte_hexdump.h rte_devargs.h rte_bus.h rte_dev.h rte_vdev.h
>  INC += rte_pci_dev_feature_defs.h rte_pci_dev_features.h
>  INC += rte_malloc.h rte_keepalive.h rte_time.h
> +INC += rte_service.h rte_service_private.h
>  
>  GENERIC_INC := rte_atomic.h rte_byteorder.h rte_cycles.h rte_prefetch.h
>  GENERIC_INC += rte_spinlock.h rte_memcpy.h rte_cpuflags.h rte_rwlock.h
> diff --git a/lib/librte_eal/common/include/rte_eal.h b/lib/librte_eal/common/include/rte_eal.h
> index abf020b..1f203f8 100644
> --- a/lib/librte_eal/common/include/rte_eal.h
> +++ b/lib/librte_eal/common/include/rte_eal.h
> @@ -61,6 +61,7 @@ extern "C" {
>  enum rte_lcore_role_t {
>  	ROLE_RTE,
>  	ROLE_OFF,
> +	ROLE_SERVICE,
>  };
>  
>  /**
> @@ -80,6 +81,7 @@ enum rte_proc_type_t {
>  struct rte_config {
>  	uint32_t master_lcore;       /**< Id of the master lcore */
>  	uint32_t lcore_count;        /**< Number of available logical cores. */
> +	uint32_t score_count;        /**< Number of available service cores. */

Should we call it as service core or service lcore?

>  	enum rte_lcore_role_t lcore_role[RTE_MAX_LCORE]; /**< State of cores. */
>  
>  	/** Primary or secondary configuration */
> @@ -185,6 +187,8 @@ int rte_eal_iopl_init(void);
>   *
>   *     EPROTO indicates that the PCI bus is either not present, or is not
>   *            readable by the eal.
> + *
> + *     ENOEXEC indicates that a service core failed to launch successfully.
>   */

> +#define RTE_SERVICE_CAP_MT_SAFE (1 << 0)
> +
> +/** Return the number of services registered.
> + *
> + * The number of services registered can be passed to *rte_service_get_by_id*,
> + * enabling the application to retireve the specificaion of each service.

s/retireve the specificaion/retrieve the specification

> + *
> + * @return The number of services registered.
> + */
> +uint32_t rte_service_get_count(void);
> +
> +/** Return the specificaion of each service.

s/specificaion/specification

> + *
> + * This function provides the specification of a service. This can be used by
> + * the application to understand what the service represents. The service
> + * must not be modified by the application directly, only passed to the various
> + * rte_service_* functions.
> + *
> + * @param id The integer id of the service to retrieve
> + * @retval non-zero A valid pointer to the service_spec
> + * @retval NULL Invalid *id* provided.
> + */
> +struct rte_service_spec *rte_service_get_by_id(uint32_t id);
> +
> +/** Return the name of the service.
> + *
> + * @return A pointer to the name of the service. The returned pointer remains
> + *         in ownership of the service, and the application must not free it.
> + */
> +const char *rte_service_get_name(const struct rte_service_spec *service);
> +
> +/* Check if a service has a specific capability.

Missing the doxygen marker(ie. change to /** Check)

> + *
> + * This function returns if *service* has implements *capability*.
> + * See RTE_SERVICE_CAP_* defines for a list of valid capabilities.
> + * @retval 1 Capability supported by this service instance
> + * @retval 0 Capability not supported by this service instance
> + */
> +int32_t rte_service_probe_capability(const struct rte_service_spec *service,
> +				     uint32_t capability);
> +
> +/* Start a service core.

Missing the doxygen marker(ie. change to /** Start)

> + *
> + * Starting a core makes the core begin polling. Any services assigned to it
> + * will be run as fast as possible.
> + *
> + * @retval 0 Success
> + * @retval -EINVAL Failed to start core. The *lcore_id* passed in is not
> + *          currently assigned to be a service core.
> + */
> +int32_t rte_service_core_start(uint32_t lcore_id);
> +
> +/* Stop a service core.

Missing the doxygen marker(ie. change to /** Stop)

> + *
> + * Stopping a core makes the core become idle, but remains  assigned as a
> + * service core.
> + *
> + * @retval 0 Success
> + * @retval -EINVAL Invalid *lcore_id* provided
> + * @retval -EALREADY Already stopped core
> + * @retval -EBUSY Failed to stop core, as it would cause a service to not
> + *          be run, as this is the only core currently running the service.
> + *          The application must stop the service first, and then stop the
> + *          lcore.
> + */
> +int32_t rte_service_core_stop(uint32_t lcore_id);
> +
> +/** Retreve the number of service cores currently avaialble.
typo: ^^^^^^^^                                      ^^^^^^^^^^
Retrieve the number of service cores currently available.

> + *
> + * This function returns the integer count of service cores available. The
> + * service core count can be used in mapping logic when creating mappings
> + * from service cores to services.
> + *
> + * See *rte_service_core_list* for details on retrieving the lcore_id of each
> + * service core.
> + *
> + * @return The number of service cores currently configured.
> + */
> +int32_t rte_service_core_count(void);
> +
> +/** Retrieve the list of currently enabled service cores.
> + *
> + * This function fills in an application supplied array, with each element
> + * indicating the lcore_id of a service core.
> + *
> + * Adding and removing service cores can be performed using
> + * *rte_service_core_add* and *rte_service_core_del*.
> + * @param array An array of at least N items.

@param [out]  array An array of at least n items

> + * @param The size of *array*.

@param n The size of *array*.

> + * @retval >=0 Number of service cores that have been populated in the array
> + * @retval -ENOMEM The provided array is not large enough to fill in the
> + *          service core list. No items have been populated, call this function
> + *          with a size of at least *rte_service_core_count* items.
> + */
> +int32_t rte_service_core_list(uint32_t array[], uint32_t n);
> +
> +/** Dumps any information available about the service. If service is NULL,
> + * dumps info for all services.
> + */
> +int32_t rte_service_dump(FILE *f, struct rte_service_spec *service);
> +
> +#ifdef __cplusplus
> +}
> +#endif
> +
> +
> +#endif /* _RTE_SERVICE_H_ */
> diff --git a/lib/librte_eal/common/include/rte_service_private.h b/lib/librte_eal/common/include/rte_service_private.h
> new file mode 100644
> index 0000000..d8bb644
> --- /dev/null
> +++ b/lib/librte_eal/common/include/rte_service_private.h
> @@ -0,0 +1,108 @@
> +/* This file specifies the internal service specification.
> + * Include this file if you are writing a component that requires CPU cycles to
> + * operate, and you wish to run the component using service cores
> + */
> +
> +#include <rte_service.h>
> +struct rte_service_spec {
> +        /** The name of the service. This should be used by the application to  
> +         * understand what purpose this service provides.
> +         */
> +        char name[RTE_SERVICE_NAME_MAX];
> +        /** The callback to invoke to run one iteration of the service */      
> +        rte_service_func callback;
> +        /** The userdata pointer provided to the service callback. */           
> +        void *callback_userdata;
> +        /** Flags to indicate the capabilities of this service. See
> + * defines in
> +         * the public header file for values of RTE_SERVICE_CAP_*               
> +        */
> +       uint32_t capabilities;
> +       /** NUMA socket ID that this service is affinitized to */               
> +       int8_t socket_id;

All other places socket_id is of type "int". I think, we can maintenance
the consistency here too. Looks like socket_id == SOCKET_ID_ANY not take
care in implementation if so, take care of it.

> +};

> +
> +int32_t rte_service_register(const struct rte_service_spec *spec);
> +
> +/** Unregister a service.
> + *
> + * The service being removed must be stopped before calling this function.
> + *
> + * @retval 0 The service was successfully unregistered.
> + * @retval -EBUSY The service is currently running, stop the service before
> + *          calling unregister. No action has been taken.
> + */
> +int32_t rte_service_unregister(struct rte_service_spec *service);
> +
> +/** Private function to allow EAL to initialied default mappings.

typo:                                   ^^^^^^^^^^^

> + *
> + * This function iterates all the services, and maps then to the available
> + * cores. Based on the capabilities of the services, they are set to run on the
> + * available cores in a round-robin manner.
> + *
> + * @retval 0 Success
> + */
> +int32_t rte_service_init_default_mapping(void);
> +
> +#endif /* _RTE_SERVICE_PRIVATE_H_ */
> diff --git a/lib/librte_eal/common/rte_service.c b/lib/librte_eal/common/rte_service.c
> new file mode 100644
> index 0000000..8b5e344
> --- /dev/null
> +++ b/lib/librte_eal/common/rte_service.c
> +#define RTE_SERVICE_NUM_MAX 64
> +
> +#define RTE_SERVICE_FLAG_REGISTERED_SHIFT 0

Internal macro, Can be shorten to reduce the length(SERVICE_F_REGISTERED?)

> +
> +#define RTE_SERVICE_RUNSTATE_STOPPED 0
> +#define RTE_SERVICE_RUNSTATE_RUNNING 1

Internal macro, Can be shorten to reduce the length(SERVICE_STATE_RUNNING?)


> +
> +/* internal representation of a service */
> +struct rte_service_spec_impl {
> +	/* public part of the struct */
> +	struct rte_service_spec spec;

Nice approach.

> +
> +	/* atomic lock that when set indicates a service core is currently
> +	 * running this service callback. When not set, a core may take the
> +	 * lock and then run the service callback.
> +	 */
> +	rte_atomic32_t execute_lock;
> +
> +	/* API set/get-able variables */
> +	int32_t runstate;
> +	uint8_t internal_flags;
> +
> +	/* per service statistics */
> +	uint32_t num_mapped_cores;
> +	uint64_t calls;
> +	uint64_t cycles_spent;
> +};

Since it been used in fastpath. better to align to cache line

> +
> +/* the internal values of a service core */
> +struct core_state {
> +	uint64_t service_mask; /* map of services IDs are run on this core */
> +	uint8_t runstate; /* running or stopped */
> +	uint8_t is_service_core; /* set if core is currently a service core */
> +
> +	/* extreme statistics */
> +	uint64_t calls_per_service[RTE_SERVICE_NUM_MAX];
> +};

aligned to cache line?

> +
> +static uint32_t rte_service_count;
> +static struct rte_service_spec_impl rte_services[RTE_SERVICE_NUM_MAX];
> +static struct core_state cores_state[RTE_MAX_LCORE];

Since these variable are used in fastpath, better to allocate form
huge page area. It will avoid lot of global variables in code as well.
Like other module, you can add a private function for service init and it can be
called from eal_init()


> +
> +/* returns 1 if service is registered and has not been unregistered
> + * Returns 0 if service never registered, or has been unregistered
> + */
> +static int

static inline int

> +service_valid(uint32_t id) {
> +	return !!(rte_services[id].internal_flags &
> +		 (1 << RTE_SERVICE_FLAG_REGISTERED_SHIFT));
> +}
> +
> +uint32_t
> +rte_service_get_count(void)
> +{
> +	return rte_service_count;
> +}
> +
> +struct rte_service_spec *
> +rte_service_get_by_id(uint32_t id)
> +{
> +	struct rte_service_spec *service = NULL;
> +	if (id < rte_service_count)
> +		service = (struct rte_service_spec *)&rte_services[id];
> +
> +	return service;
> +}
> +
> +const char *
> +rte_service_get_name(const struct rte_service_spec *service)
> +{
> +	return service->name;
> +}
> +
> +int32_t

bool could be enough here

> +rte_service_probe_capability(const struct rte_service_spec *service,
> +			     uint32_t capability)
> +{
> +	return service->capabilities & capability;
> +}
> +
> +int32_t
> +rte_service_is_running(const struct rte_service_spec *spec)
> +{
> +	if (!spec)
> +		return -EINVAL;
> +
> +	const struct rte_service_spec_impl *impl =
> +		(const struct rte_service_spec_impl *)spec;
> +	return impl->runstate == RTE_SERVICE_RUNSTATE_RUNNING;
> +}
> +
> +int32_t
> +rte_service_register(const struct rte_service_spec *spec)
> +{
> +	uint32_t i;
> +	int32_t free_slot = -1;
> +
> +	if (spec->callback == NULL || strlen(spec->name) == 0)
> +		return -EINVAL;
> +
> +	for (i = 0; i < RTE_SERVICE_NUM_MAX; i++) {
> +		if (!service_valid(i)) {
> +			free_slot = i;
> +			break;
> +		}
> +	}
> +
> +	if (free_slot < 0)

	if ((free_slot < 0) || (i == RTE_SERVICE_NUM_MAX))

> +		return -ENOSPC;
> +
> +	struct rte_service_spec_impl *s = &rte_services[free_slot];
> +	s->spec = *spec;
> +	s->internal_flags |= (1 << RTE_SERVICE_FLAG_REGISTERED_SHIFT);
> +
> +	rte_smp_wmb();
> +	rte_service_count++;

IMO, You can move above rte_smp_wmb() here.

> +
> +	return 0;
> +}
> +
> +int32_t
> +rte_service_unregister(struct rte_service_spec *spec)
> +{
> +	struct rte_service_spec_impl *s = NULL;
> +	struct rte_service_spec_impl *spec_impl =
> +		(struct rte_service_spec_impl *)spec;
> +
> +	uint32_t i;
> +	uint32_t service_id;
> +	for (i = 0; i < RTE_SERVICE_NUM_MAX; i++) {
> +		if (&rte_services[i] == spec_impl) {
> +			s = spec_impl;
> +			service_id = i;
> +			break;
> +		}
> +	}
> +
> +	if (!s)
> +		return -EINVAL;
> +
> +	s->internal_flags &= ~(1 << RTE_SERVICE_FLAG_REGISTERED_SHIFT);
> +
> +	for (i = 0; i < RTE_MAX_LCORE; i++)
> +		cores_state[i].service_mask &= ~(1 << service_id);
> +
> +	memset(&rte_services[service_id], 0,
> +			sizeof(struct rte_service_spec_impl));
> +
> +	rte_smp_wmb();
> +	rte_service_count--;

IMO, You can move above rte_smp_wmb() here.

> +
> +	return 0;
> +}
> +
> +int32_t
> +rte_service_start(struct rte_service_spec *service)
> +{
> +	struct rte_service_spec_impl *s =
> +		(struct rte_service_spec_impl *)service;
> +	s->runstate = RTE_SERVICE_RUNSTATE_RUNNING;

Is this function can called from worker thread? if so add rte_smp_wmb()

> +	return 0;
> +}
> +
> +int32_t
> +rte_service_stop(struct rte_service_spec *service)
> +{
> +	struct rte_service_spec_impl *s =
> +		(struct rte_service_spec_impl *)service;
> +	s->runstate = RTE_SERVICE_RUNSTATE_STOPPED;

Is this function can called from worker thread? if so add rte_smp_wmb()

> +	return 0;
> +}
> +
> +static int32_t
> +rte_service_runner_func(void *arg)
> +{
> +	RTE_SET_USED(arg);
> +	uint32_t i;
> +	const int lcore = rte_lcore_id();
> +	struct core_state *cs = &cores_state[lcore];
> +
> +	while (cores_state[lcore].runstate == RTE_SERVICE_RUNSTATE_RUNNING) {
> +		for (i = 0; i < rte_service_count; i++) {
> +			struct rte_service_spec_impl *s = &rte_services[i];
> +			uint64_t service_mask = cs->service_mask;

No need to read in loop, Move it above while loop and add const.
const uint64_t service_mask = cs->service_mask;


> +
> +			if (s->runstate != RTE_SERVICE_RUNSTATE_RUNNING ||
> +					!(service_mask & (1 << i)))
> +				continue;
> +
> +			uint32_t *lock = (uint32_t *)&s->execute_lock;
> +			if (rte_atomic32_cmpset(lock, 0, 1)) {

rte_atomic32 is costly. How about checking RTE_SERVICE_CAP_MT_SAFE
first.

> +				void *userdata = s->spec.callback_userdata;
> +				uint64_t start = rte_rdtsc();
> +				s->spec.callback(userdata);
> +				uint64_t end = rte_rdtsc();
> +
> +				uint64_t spent = end - start;
> +				s->cycles_spent += spent;
> +				s->calls++;
> +				cs->calls_per_service[i]++;

How about enabling the statistics based on some runtime configuration?

> +
> +				rte_atomic32_clear(&s->execute_lock);
> +			}
> +		}
> +		rte_mb();

Do we need full barrier here. Is rte_smp_rmb() inside the loop is
enough?

> +	}
> +
> +	/* mark core as ready to accept work again */
> +	lcore_config[lcore].state = WAIT;
> +
> +	return 0;
> +}
> +
> +int32_t
> +rte_service_core_count(void)
> +{
> +	int32_t count = 0;
> +	uint32_t i;
> +	for (i = 0; i < RTE_MAX_LCORE; i++)
> +		count += cores_state[i].is_service_core;
> +	return count;
> +}
> +
> +int32_t
> +rte_service_core_list(uint32_t array[], uint32_t n)
> +{
> +	uint32_t count = rte_service_core_count();

	if (!array)
		return -EINVAL;

> +	if (count > n)
> +		return -ENOMEM;
> +
> +	uint32_t i;
> +	uint32_t idx = 0;
> +	for (i = 0; i < RTE_MAX_LCORE; i++) {

Are we good if "count" being the upper limit instead of RTE_MAX_LCORE?

> +		struct core_state *cs = &cores_state[i];
> +		if (cs->is_service_core) {
> +			array[idx] = i;
> +			idx++;
> +		}
> +	}
> +
> +	return count;
> +}
> +
> +int32_t
> +rte_service_init_default_mapping(void)
> +{
> +	/* create a default mapping from cores to services, then start the
> +	 * services to make them transparent to unaware applications.
> +	 */
> +	uint32_t i;
> +	int ret;
> +	uint32_t count = rte_service_get_count();
> +	struct rte_config *cfg = rte_eal_get_configuration();
> +
> +	for (i = 0; i < count; i++) {
> +		struct rte_service_spec *s = rte_service_get_by_id(i);
> +		if (!s)
> +			return -EINVAL;
> +
> +		ret = 0;
> +		int j;
> +		for (j = 0; j < RTE_MAX_LCORE; j++) {
> +			/* TODO: add lcore -> service mapping logic here */
> +			if (cfg->lcore_role[j] == ROLE_SERVICE) {
> +				ret = rte_service_enable_on_core(s, j);
> +				if (ret)
> +					rte_panic("Enabling service core %d on service %s failed\n",
> +							j, s->name);

avoid panic in library


> +			}
> +		}
> +
> +		ret = rte_service_start(s);
> +		if (ret)
> +			rte_panic("failed to start service %s\n", s->name);

avoid panic in library

> +	}
> +
> +	return 0;
> +}
> +
> +static int32_t
> +service_update(struct rte_service_spec *service, uint32_t lcore,
> +		uint32_t *set, uint32_t *enabled)
> +{
> +	uint32_t i;
> +	int32_t sid = -1;
> +
> +	for (i = 0; i < RTE_SERVICE_NUM_MAX; i++) {
> +		if ((struct rte_service_spec *)&rte_services[i] == service &&
> +				service_valid(i)) {
> +			sid = i;
> +			break;
> +		}
> +	}
> +
> +	if (sid == -1 || lcore >= RTE_MAX_LCORE)
> +		return -EINVAL;
> +
> +	if (!cores_state[lcore].is_service_core)
> +		return -EINVAL;
> +
> +	if (set) {
> +		if (*set) {
> +			cores_state[lcore].service_mask |=  (1 << sid);
> +			rte_services[sid].num_mapped_cores++;
> +		} else {
> +			cores_state[lcore].service_mask &= ~(1 << sid);
> +			rte_services[sid].num_mapped_cores--;
> +		}
> +	}
> +
> +	if (enabled)
> +		*enabled = (cores_state[lcore].service_mask & (1 << sid));

If the parent functions can be called from worker thread then add
rte_smp_wmb() here.


> +
> +	return 0;
> +}
> +
> +int32_t rte_service_get_enabled_on_core(struct rte_service_spec *service,
> +					uint32_t lcore)
> +{
> +	uint32_t enabled;
> +	int ret = service_update(service, lcore, 0, &enabled);
> +	if (ret == 0)
> +		return enabled;
> +	return -EINVAL;
> +}
> +
> +int32_t
> +rte_service_enable_on_core(struct rte_service_spec *service, uint32_t lcore)
> +{
> +	uint32_t on = 1;
> +	return service_update(service, lcore, &on, 0);
> +}
> +
> +int32_t
> +rte_service_disable_on_core(struct rte_service_spec *service, uint32_t lcore)
> +{
> +	uint32_t off = 0;
> +	return service_update(service, lcore, &off, 0);
> +}
> +
> +int32_t rte_service_core_reset_all(void)
> +{
> +	/* loop over cores, reset all to mask 0 */
> +	uint32_t i;
> +	for (i = 0; i < RTE_MAX_LCORE; i++) {
> +		cores_state[i].service_mask = 0;
> +		cores_state[i].is_service_core = 0;
> +	}
> +
> +	return 0;
> +}
> +
> +int32_t
> +rte_service_core_add(uint32_t lcore)
> +{
> +	if (lcore >= RTE_MAX_LCORE)
> +		return -EINVAL;
> +	if (cores_state[lcore].is_service_core)
> +		return -EALREADY;
> +
> +	lcore_config[lcore].core_role = ROLE_SERVICE;
> +
> +	/* TODO: take from EAL by setting ROLE_SERVICE? */

I think, we need to fix TODO in v2

> +	cores_state[lcore].is_service_core = 1;
> +	cores_state[lcore].service_mask = 0;
> +
> +	return 0;
> +}
> +
> +int32_t
> +rte_service_core_del(uint32_t lcore)
> +{
> +	if (lcore >= RTE_MAX_LCORE)
> +		return -EINVAL;
> +
> +	struct core_state *cs = &cores_state[lcore];
> +	if (!cs->is_service_core)
> +		return -EINVAL;
> +
> +	if (cs->runstate != RTE_SERVICE_RUNSTATE_STOPPED)
> +		return -EBUSY;
> +
> +	lcore_config[lcore].core_role = ROLE_RTE;
> +	cores_state[lcore].is_service_core = 0;
> +	/* TODO: return to EAL by setting ROLE_RTE? */

I think, we need to fix TODO in v2

> +
> +	return 0;
> +}
> +
> +int32_t
> +rte_service_core_start(uint32_t lcore)
> +{
> +	if (lcore >= RTE_MAX_LCORE)
> +		return -EINVAL;
> +
> +	struct core_state *cs = &cores_state[lcore];
> +	if (!cs->is_service_core)
> +		return -EINVAL;
> +
> +	if (cs->runstate == RTE_SERVICE_RUNSTATE_RUNNING)
> +		return -EALREADY;
> +
> +	/* set core to run state first, and then launch otherwise it will
> +	 * return immidiatly as runstate keeps it in the service poll loop

s/immidiatly/immediately

> +	 */
> +	cores_state[lcore].runstate = RTE_SERVICE_RUNSTATE_RUNNING;
> +
> +	int ret = rte_eal_remote_launch(rte_service_runner_func, 0, lcore);
> +	/* returns -EBUSY if the core is already launched, 0 on success */
> +	return ret;

return rte_eal_remote_launch(rte_service_runner_func, 0, lcore);


> +}
> +
> +int32_t
> +rte_service_core_stop(uint32_t lcore)
> +{
> +	if (lcore >= RTE_MAX_LCORE)
> +		return -EINVAL;
> +
> +	if (cores_state[lcore].runstate == RTE_SERVICE_RUNSTATE_STOPPED)
> +		return -EALREADY;
> +
> +	uint32_t i;
> +	for (i = 0; i < RTE_SERVICE_NUM_MAX; i++) {
> +		int32_t enabled = cores_state[i].service_mask & (1 << i);
> +		int32_t service_running = rte_services[i].runstate !=
> +						RTE_SERVICE_RUNSTATE_STOPPED;
> +		int32_t only_core = rte_services[i].num_mapped_cores == 1;
> +
> +		/* if the core is mapped, and the service is running, and this
> +		 * is the only core that is mapped, the service would cease to
> +		 * run if this core stopped, so fail instead.
> +		 */
> +		if (enabled && service_running && only_core)
> +			return -EBUSY;
> +	}
> +
> +	cores_state[lcore].runstate = RTE_SERVICE_RUNSTATE_STOPPED;
> +
> +	return 0;
> +}
> +
> +static void
> +rte_service_dump_one(FILE *f, struct rte_service_spec_impl *s,
> +		     uint64_t all_cycles, uint32_t reset)
> +{
> +	/* avoid divide by zeros */

s/zeros/zero

> +	if (all_cycles == 0)
> +		all_cycles = 1;
> +
> +	int calls = 1;
> +	if (s->calls != 0)
> +		calls = s->calls;
> +
> +	float cycles_pct = (((float)s->cycles_spent) / all_cycles) * 100.f;
> +	fprintf(f,
> +			"  %s : %0.1f %%\tcalls %"PRIu64"\tcycles %"PRIu64"\tavg: %"PRIu64"\n",
> +			s->spec.name, cycles_pct, s->calls, s->cycles_spent,
> +			s->cycles_spent / calls);
> +
> +	if (reset) {
> +		s->cycles_spent = 0;
> +		s->calls = 0;
> +	}
> +}
> +