X-Git-Url: http://git.droids-corp.org/?a=blobdiff_plain;f=lib%2Flibrte_eal%2Fcommon%2Feal_common_proc.c;h=6d1af3c0e756a2975bb1390a5770b916daaf75ca;hb=e863fe3a13da89787fdf3b5c590101a3c0f10af6;hp=f98622f4e9be5077e8b4c5b2f5a3730b8187a6c4;hpb=f05e26051c15dd208277c7e27c77f67720d5d4b1;p=dpdk.git diff --git a/lib/librte_eal/common/eal_common_proc.c b/lib/librte_eal/common/eal_common_proc.c index f98622f4e9..6d1af3c0e7 100644 --- a/lib/librte_eal/common/eal_common_proc.c +++ b/lib/librte_eal/common/eal_common_proc.c @@ -20,6 +20,7 @@ #include #include +#include #include #include #include @@ -28,6 +29,7 @@ #include #include +#include "eal_memcfg.h" #include "eal_private.h" #include "eal_filesystem.h" #include "eal_internal_cfg.h" @@ -36,6 +38,7 @@ static int mp_fd = -1; static char mp_filter[PATH_MAX]; /* Filter for secondary process sockets */ static char mp_dir_path[PATH_MAX]; /* The directory path for all mp sockets */ static pthread_mutex_t mp_mutex_action = PTHREAD_MUTEX_INITIALIZER; +static char peer_name[PATH_MAX]; struct action_entry { TAILQ_ENTRY(action_entry) next; @@ -94,11 +97,9 @@ TAILQ_HEAD(pending_request_list, pending_request); static struct { struct pending_request_list requests; pthread_mutex_t lock; - pthread_cond_t async_cond; } pending_requests = { .requests = TAILQ_HEAD_INITIALIZER(pending_requests.requests), .lock = PTHREAD_MUTEX_INITIALIZER, - .async_cond = PTHREAD_COND_INITIALIZER /**< used in async requests only */ }; @@ -106,9 +107,19 @@ static struct { static int mp_send(struct rte_mp_msg *msg, const char *peer, int type); +/* for use with alarm callback */ +static void +async_reply_handle(void *arg); + +/* for use with process_msg */ +static struct pending_request * +async_reply_handle_thread_unsafe(void *arg); + +static void +trigger_async_action(struct pending_request *req); static struct pending_request * -find_sync_request(const char *dst, const char *act_name) +find_pending_request(const char *dst, const char *act_name) { struct pending_request *r; @@ -129,7 +140,7 @@ create_socket_path(const char *name, char *buf, int len) if (strlen(name) > 0) snprintf(buf, len, "%s_%s", prefix, name); else - snprintf(buf, len, "%s", prefix); + strlcpy(buf, prefix, len); } int @@ -187,20 +198,28 @@ validate_action_name(const char *name) return 0; } -int __rte_experimental +int rte_mp_action_register(const char *name, rte_mp_t action) { struct action_entry *entry; + const struct internal_config *internal_conf = + eal_get_internal_configuration(); - if (validate_action_name(name)) + if (validate_action_name(name) != 0) return -1; + if (internal_conf->no_shconf) { + RTE_LOG(DEBUG, EAL, "No shared files mode enabled, IPC is disabled\n"); + rte_errno = ENOTSUP; + return -1; + } + entry = malloc(sizeof(struct action_entry)); if (entry == NULL) { rte_errno = ENOMEM; return -1; } - strcpy(entry->action_name, name); + strlcpy(entry->action_name, name, sizeof(entry->action_name)); entry->action = action; pthread_mutex_lock(&mp_mutex_action); @@ -215,14 +234,21 @@ rte_mp_action_register(const char *name, rte_mp_t action) return 0; } -void __rte_experimental +void rte_mp_action_unregister(const char *name) { struct action_entry *entry; + const struct internal_config *internal_conf = + eal_get_internal_configuration(); - if (validate_action_name(name)) + if (validate_action_name(name) != 0) return; + if (internal_conf->no_shconf) { + RTE_LOG(DEBUG, EAL, "No shared files mode enabled, IPC is disabled\n"); + return; + } + pthread_mutex_lock(&mp_mutex_action); entry = find_action_entry_by_name(name); if (entry == NULL) { @@ -262,7 +288,7 @@ read_msg(struct mp_msg_internal *m, struct sockaddr_un *s) } if (msglen != buflen || (msgh.msg_flags & (MSG_TRUNC | MSG_CTRUNC))) { - RTE_LOG(ERR, EAL, "truncted msg\n"); + RTE_LOG(ERR, EAL, "truncated msg\n"); return -1; } @@ -275,36 +301,52 @@ read_msg(struct mp_msg_internal *m, struct sockaddr_un *s) break; } } - + /* sanity-check the response */ + if (m->msg.num_fds < 0 || m->msg.num_fds > RTE_MP_MAX_FD_NUM) { + RTE_LOG(ERR, EAL, "invalid number of fd's received\n"); + return -1; + } + if (m->msg.len_param < 0 || m->msg.len_param > RTE_MP_MAX_PARAM_LEN) { + RTE_LOG(ERR, EAL, "invalid received data length\n"); + return -1; + } return 0; } static void process_msg(struct mp_msg_internal *m, struct sockaddr_un *s) { - struct pending_request *sync_req; + struct pending_request *pending_req; struct action_entry *entry; struct rte_mp_msg *msg = &m->msg; rte_mp_t action = NULL; + const struct internal_config *internal_conf = + eal_get_internal_configuration(); RTE_LOG(DEBUG, EAL, "msg: %s\n", msg->name); if (m->type == MP_REP || m->type == MP_IGN) { + struct pending_request *req = NULL; + pthread_mutex_lock(&pending_requests.lock); - sync_req = find_sync_request(s->sun_path, msg->name); - if (sync_req) { - memcpy(sync_req->reply, msg, sizeof(*msg)); + pending_req = find_pending_request(s->sun_path, msg->name); + if (pending_req) { + memcpy(pending_req->reply, msg, sizeof(*msg)); /* -1 indicates that we've been asked to ignore */ - sync_req->reply_received = m->type == MP_REP ? 1 : -1; - - if (sync_req->type == REQUEST_TYPE_SYNC) - pthread_cond_signal(&sync_req->sync.cond); - else if (sync_req->type == REQUEST_TYPE_ASYNC) - pthread_cond_signal( - &pending_requests.async_cond); + pending_req->reply_received = + m->type == MP_REP ? 1 : -1; + + if (pending_req->type == REQUEST_TYPE_SYNC) + pthread_cond_signal(&pending_req->sync.cond); + else if (pending_req->type == REQUEST_TYPE_ASYNC) + req = async_reply_handle_thread_unsafe( + pending_req); } else RTE_LOG(ERR, EAL, "Drop mp reply: %s\n", msg->name); pthread_mutex_unlock(&pending_requests.lock); + + if (req != NULL) + trigger_async_action(req); return; } @@ -315,14 +357,16 @@ process_msg(struct mp_msg_internal *m, struct sockaddr_un *s) pthread_mutex_unlock(&mp_mutex_action); if (!action) { - if (m->type == MP_REQ && !internal_config.init_complete) { + if (m->type == MP_REQ && !internal_conf->init_complete) { /* if this is a request, and init is not yet complete, * and callback wasn't registered, we should tell the * requester to ignore our existence because we're not * yet ready to process this request. */ struct rte_mp_msg dummy; + memset(&dummy, 0, sizeof(dummy)); + strlcpy(dummy.name, msg->name, sizeof(dummy.name)); mp_send(&dummy, s->sun_path, MP_IGN); } else { RTE_LOG(ERR, EAL, "Cannot find action: %s\n", @@ -362,7 +406,6 @@ timespec_cmp(const struct timespec *a, const struct timespec *b) } enum async_action { - ACTION_NONE, /**< don't do anything */ ACTION_FREE, /**< free the action entry, but don't trigger callback */ ACTION_TRIGGER /**< trigger callback, then free action entry */ }; @@ -372,7 +415,7 @@ process_async_request(struct pending_request *sr, const struct timespec *now) { struct async_request_param *param; struct rte_mp_reply *reply; - bool timeout, received, last_msg; + bool timeout, last_msg; param = sr->async.param; reply = ¶m->user_reply; @@ -380,14 +423,7 @@ process_async_request(struct pending_request *sr, const struct timespec *now) /* did we timeout? */ timeout = timespec_cmp(¶m->end, now) <= 0; - /* did we receive a response? */ - received = sr->reply_received != 0; - - /* if we didn't time out, and we didn't receive a response, ignore */ - if (!timeout && !received) - return ACTION_NONE; - - /* if we received a response, adjust relevant data and copy mesasge. */ + /* if we received a response, adjust relevant data and copy message. */ if (sr->reply_received == 1 && sr->reply) { struct rte_mp_msg *msg, *user_msgs, *tmp; @@ -416,7 +452,13 @@ process_async_request(struct pending_request *sr, const struct timespec *now) } else if (sr->reply_received == -1) { /* we were asked to ignore this process */ reply->nb_sent--; + } else if (timeout) { + /* count it as processed response, but don't increment + * nb_received. + */ + param->n_responses_processed++; } + free(sr->reply); last_msg = param->n_responses_processed == reply->nb_sent; @@ -439,103 +481,66 @@ trigger_async_action(struct pending_request *sr) free(sr->async.param->user_reply.msgs); free(sr->async.param); free(sr->request); + free(sr); } -static void * -async_reply_handle(void *arg __rte_unused) +static struct pending_request * +async_reply_handle_thread_unsafe(void *arg) { - struct pending_request *sr; + struct pending_request *req = (struct pending_request *)arg; + enum async_action action; + struct timespec ts_now; struct timeval now; - struct timespec timeout, ts_now; - while (1) { - struct pending_request *trigger = NULL; - int ret; - bool nowait = false; - bool timedwait = false; - pthread_mutex_lock(&pending_requests.lock); + if (gettimeofday(&now, NULL) < 0) { + RTE_LOG(ERR, EAL, "Cannot get current time\n"); + goto no_trigger; + } + ts_now.tv_nsec = now.tv_usec * 1000; + ts_now.tv_sec = now.tv_sec; - /* scan through the list and see if there are any timeouts that - * are earlier than our current timeout. - */ - TAILQ_FOREACH(sr, &pending_requests.requests, next) { - if (sr->type != REQUEST_TYPE_ASYNC) - continue; - if (!timedwait || timespec_cmp(&sr->async.param->end, - &timeout) < 0) { - memcpy(&timeout, &sr->async.param->end, - sizeof(timeout)); - timedwait = true; - } - - /* sometimes, we don't even wait */ - if (sr->reply_received) { - nowait = true; - break; - } - } + action = process_async_request(req, &ts_now); - if (nowait) - ret = 0; - else if (timedwait) - ret = pthread_cond_timedwait( - &pending_requests.async_cond, - &pending_requests.lock, &timeout); - else - ret = pthread_cond_wait(&pending_requests.async_cond, - &pending_requests.lock); + TAILQ_REMOVE(&pending_requests.requests, req, next); - if (gettimeofday(&now, NULL) < 0) { - RTE_LOG(ERR, EAL, "Cannot get current time\n"); - break; + if (rte_eal_alarm_cancel(async_reply_handle, req) < 0) { + /* if we failed to cancel the alarm because it's already in + * progress, don't proceed because otherwise we will end up + * handling the same message twice. + */ + if (rte_errno == EINPROGRESS) { + RTE_LOG(DEBUG, EAL, "Request handling is already in progress\n"); + goto no_trigger; } - ts_now.tv_nsec = now.tv_usec * 1000; - ts_now.tv_sec = now.tv_sec; + RTE_LOG(ERR, EAL, "Failed to cancel alarm\n"); + } - if (ret == 0 || ret == ETIMEDOUT) { - struct pending_request *next; - /* we've either been woken up, or we timed out */ + if (action == ACTION_TRIGGER) + return req; +no_trigger: + free(req); + return NULL; +} - /* we have still the lock, check if anything needs - * processing. - */ - TAILQ_FOREACH_SAFE(sr, &pending_requests.requests, next, - next) { - enum async_action action; - if (sr->type != REQUEST_TYPE_ASYNC) - continue; - - action = process_async_request(sr, &ts_now); - if (action == ACTION_FREE) { - TAILQ_REMOVE(&pending_requests.requests, - sr, next); - free(sr); - } else if (action == ACTION_TRIGGER && - trigger == NULL) { - TAILQ_REMOVE(&pending_requests.requests, - sr, next); - trigger = sr; - } - } - } - pthread_mutex_unlock(&pending_requests.lock); - if (trigger) { - trigger_async_action(trigger); - free(trigger); - } - }; +static void +async_reply_handle(void *arg) +{ + struct pending_request *req; - RTE_LOG(ERR, EAL, "ERROR: asynchronous requests disabled\n"); + pthread_mutex_lock(&pending_requests.lock); + req = async_reply_handle_thread_unsafe(arg); + pthread_mutex_unlock(&pending_requests.lock); - return NULL; + if (req != NULL) + trigger_async_action(req); } static int open_socket_fd(void) { - char peer_name[PATH_MAX] = {0}; struct sockaddr_un un; + peer_name[0] = '\0'; if (rte_eal_process_type() == RTE_PROC_SECONDARY) snprintf(peer_name, sizeof(peer_name), "%d_%"PRIx64, getpid(), rte_rdtsc()); @@ -564,44 +569,44 @@ open_socket_fd(void) return mp_fd; } -static int -unlink_sockets(const char *filter) +static void +close_socket_fd(void) { - int dir_fd; - DIR *mp_dir; - struct dirent *ent; - - mp_dir = opendir(mp_dir_path); - if (!mp_dir) { - RTE_LOG(ERR, EAL, "Unable to open directory %s\n", mp_dir_path); - return -1; - } - dir_fd = dirfd(mp_dir); + char path[PATH_MAX]; - while ((ent = readdir(mp_dir))) { - if (fnmatch(filter, ent->d_name, 0) == 0) - unlinkat(dir_fd, ent->d_name, 0); - } + if (mp_fd < 0) + return; - closedir(mp_dir); - return 0; + close(mp_fd); + create_socket_path(peer_name, path, sizeof(path)); + unlink(path); } int rte_mp_channel_init(void) { - char thread_name[RTE_MAX_THREAD_NAME_LEN]; char path[PATH_MAX]; int dir_fd; - pthread_t mp_handle_tid, async_reply_handle_tid; + pthread_t mp_handle_tid; + const struct internal_config *internal_conf = + eal_get_internal_configuration(); + + /* in no shared files mode, we do not have secondary processes support, + * so no need to initialize IPC. + */ + if (internal_conf->no_shconf) { + RTE_LOG(DEBUG, EAL, "No shared files mode enabled, IPC will be disabled\n"); + rte_errno = ENOTSUP; + return -1; + } /* create filter path */ create_socket_path("*", path, sizeof(path)); - snprintf(mp_filter, sizeof(mp_filter), "%s", basename(path)); + strlcpy(mp_filter, basename(path), sizeof(mp_filter)); /* path may have been modified, so recreate it */ create_socket_path("*", path, sizeof(path)); - snprintf(mp_dir_path, sizeof(mp_dir_path), "%s", dirname(path)); + strlcpy(mp_dir_path, dirname(path), sizeof(mp_dir_path)); /* lock the directory */ dir_fd = open(mp_dir_path, O_RDONLY); @@ -618,29 +623,14 @@ rte_mp_channel_init(void) return -1; } - if (rte_eal_process_type() == RTE_PROC_PRIMARY && - unlink_sockets(mp_filter)) { - RTE_LOG(ERR, EAL, "failed to unlink mp sockets\n"); - close(dir_fd); - return -1; - } - if (open_socket_fd() < 0) { close(dir_fd); return -1; } - if (pthread_create(&mp_handle_tid, NULL, mp_handle, NULL) < 0) { - RTE_LOG(ERR, EAL, "failed to create mp thead: %s\n", - strerror(errno)); - close(mp_fd); - mp_fd = -1; - return -1; - } - - if (pthread_create(&async_reply_handle_tid, NULL, - async_reply_handle, NULL) < 0) { - RTE_LOG(ERR, EAL, "failed to create mp thead: %s\n", + if (rte_ctrl_thread_create(&mp_handle_tid, "rte_mp_handle", + NULL, mp_handle, NULL) < 0) { + RTE_LOG(ERR, EAL, "failed to create mp thread: %s\n", strerror(errno)); close(mp_fd); close(dir_fd); @@ -648,14 +638,6 @@ rte_mp_channel_init(void) return -1; } - /* try best to set thread name */ - snprintf(thread_name, RTE_MAX_THREAD_NAME_LEN, "rte_mp_handle"); - rte_thread_setname(mp_handle_tid, thread_name); - - /* try best to set thread name */ - snprintf(thread_name, RTE_MAX_THREAD_NAME_LEN, "rte_mp_async_handle"); - rte_thread_setname(async_reply_handle_tid, thread_name); - /* unlock the directory */ flock(dir_fd, LOCK_UN); close(dir_fd); @@ -663,6 +645,12 @@ rte_mp_channel_init(void) return 0; } +void +rte_mp_channel_cleanup(void) +{ + close_socket_fd(); +} + /** * Return -1, as fail to send message and it's caused by the local side. * Return 0, as fail to send message and it's caused by the remote side. @@ -686,7 +674,7 @@ send_msg(const char *dst_path, struct rte_mp_msg *msg, int type) memset(&dst, 0, sizeof(dst)); dst.sun_family = AF_UNIX; - snprintf(dst.sun_path, sizeof(dst.sun_path), "%s", dst_path); + strlcpy(dst.sun_path, dst_path, sizeof(dst.sun_path)); memset(&msgh, 0, sizeof(msgh)); memset(control, 0, sizeof(control)); @@ -719,11 +707,6 @@ send_msg(const char *dst_path, struct rte_mp_msg *msg, int type) unlink(dst_path); return 0; } - if (errno == ENOBUFS) { - RTE_LOG(ERR, EAL, "Peer cannot receive message %s\n", - dst_path); - return 0; - } RTE_LOG(ERR, EAL, "failed to send to (%s) due to %s\n", dst_path, strerror(errno)); return -1; @@ -760,7 +743,7 @@ mp_send(struct rte_mp_msg *msg, const char *peer, int type) dir_fd = dirfd(mp_dir); /* lock the directory to prevent processes spinning up while we send */ - if (flock(dir_fd, LOCK_EX)) { + if (flock(dir_fd, LOCK_SH)) { RTE_LOG(ERR, EAL, "Unable to lock directory %s\n", mp_dir_path); rte_errno = errno; @@ -787,76 +770,92 @@ mp_send(struct rte_mp_msg *msg, const char *peer, int type) return ret; } -static bool +static int check_input(const struct rte_mp_msg *msg) { if (msg == NULL) { RTE_LOG(ERR, EAL, "Msg cannot be NULL\n"); rte_errno = EINVAL; - return false; + return -1; + } + + if (validate_action_name(msg->name) != 0) + return -1; + + if (msg->len_param < 0) { + RTE_LOG(ERR, EAL, "Message data length is negative\n"); + rte_errno = EINVAL; + return -1; } - if (validate_action_name(msg->name)) - return false; + if (msg->num_fds < 0) { + RTE_LOG(ERR, EAL, "Number of fd's is negative\n"); + rte_errno = EINVAL; + return -1; + } if (msg->len_param > RTE_MP_MAX_PARAM_LEN) { RTE_LOG(ERR, EAL, "Message data is too long\n"); rte_errno = E2BIG; - return false; + return -1; } if (msg->num_fds > RTE_MP_MAX_FD_NUM) { RTE_LOG(ERR, EAL, "Cannot send more than %d FDs\n", RTE_MP_MAX_FD_NUM); rte_errno = E2BIG; - return false; + return -1; } - return true; + return 0; } -int __rte_experimental +int rte_mp_sendmsg(struct rte_mp_msg *msg) { - if (!check_input(msg)) + const struct internal_config *internal_conf = + eal_get_internal_configuration(); + + if (check_input(msg) != 0) return -1; + if (internal_conf->no_shconf) { + RTE_LOG(DEBUG, EAL, "No shared files mode enabled, IPC is disabled\n"); + rte_errno = ENOTSUP; + return -1; + } + RTE_LOG(DEBUG, EAL, "sendmsg: %s\n", msg->name); return mp_send(msg, NULL, MP_MSG); } static int mp_request_async(const char *dst, struct rte_mp_msg *req, - struct async_request_param *param) + struct async_request_param *param, const struct timespec *ts) { struct rte_mp_msg *reply_msg; - struct pending_request *sync_req, *exist; - int ret; + struct pending_request *pending_req, *exist; + int ret = -1; - sync_req = malloc(sizeof(*sync_req)); - reply_msg = malloc(sizeof(*reply_msg)); - if (sync_req == NULL || reply_msg == NULL) { + pending_req = calloc(1, sizeof(*pending_req)); + reply_msg = calloc(1, sizeof(*reply_msg)); + if (pending_req == NULL || reply_msg == NULL) { RTE_LOG(ERR, EAL, "Could not allocate space for sync request\n"); rte_errno = ENOMEM; ret = -1; goto fail; } - memset(sync_req, 0, sizeof(*sync_req)); - memset(reply_msg, 0, sizeof(*reply_msg)); - - sync_req->type = REQUEST_TYPE_ASYNC; - strcpy(sync_req->dst, dst); - sync_req->request = req; - sync_req->reply = reply_msg; - sync_req->async.param = param; + pending_req->type = REQUEST_TYPE_ASYNC; + strlcpy(pending_req->dst, dst, sizeof(pending_req->dst)); + pending_req->request = req; + pending_req->reply = reply_msg; + pending_req->async.param = param; /* queue already locked by caller */ - exist = find_sync_request(dst, req->name); - if (!exist) { - TAILQ_INSERT_TAIL(&pending_requests.requests, sync_req, next); - } else { + exist = find_pending_request(dst, req->name); + if (exist) { RTE_LOG(ERR, EAL, "A pending request %s:%s\n", dst, req->name); rte_errno = EEXIST; ret = -1; @@ -873,12 +872,21 @@ mp_request_async(const char *dst, struct rte_mp_msg *req, ret = 0; goto fail; } - param->user_reply.nb_sent++; + /* if alarm set fails, we simply ignore the reply */ + if (rte_eal_alarm_set(ts->tv_sec * 1000000 + ts->tv_nsec / 1000, + async_reply_handle, pending_req) < 0) { + RTE_LOG(ERR, EAL, "Fail to set alarm for request %s:%s\n", + dst, req->name); + ret = -1; + goto fail; + } + TAILQ_INSERT_TAIL(&pending_requests.requests, pending_req, next); + return 0; fail: - free(sync_req); + free(pending_req); free(reply_msg); return ret; } @@ -889,23 +897,19 @@ mp_request_sync(const char *dst, struct rte_mp_msg *req, { int ret; struct rte_mp_msg msg, *tmp; - struct pending_request sync_req, *exist; + struct pending_request pending_req, *exist; - sync_req.type = REQUEST_TYPE_SYNC; - sync_req.reply_received = 0; - strcpy(sync_req.dst, dst); - sync_req.request = req; - sync_req.reply = &msg; - pthread_cond_init(&sync_req.sync.cond, NULL); + pending_req.type = REQUEST_TYPE_SYNC; + pending_req.reply_received = 0; + strlcpy(pending_req.dst, dst, sizeof(pending_req.dst)); + pending_req.request = req; + pending_req.reply = &msg; + pthread_cond_init(&pending_req.sync.cond, NULL); - pthread_mutex_lock(&pending_requests.lock); - exist = find_sync_request(dst, req->name); - if (!exist) - TAILQ_INSERT_TAIL(&pending_requests.requests, &sync_req, next); + exist = find_pending_request(dst, req->name); if (exist) { RTE_LOG(ERR, EAL, "A pending request %s:%s\n", dst, req->name); rte_errno = EEXIST; - pthread_mutex_unlock(&pending_requests.lock); return -1; } @@ -917,24 +921,24 @@ mp_request_sync(const char *dst, struct rte_mp_msg *req, } else if (ret == 0) return 0; + TAILQ_INSERT_TAIL(&pending_requests.requests, &pending_req, next); + reply->nb_sent++; do { - ret = pthread_cond_timedwait(&sync_req.sync.cond, + ret = pthread_cond_timedwait(&pending_req.sync.cond, &pending_requests.lock, ts); } while (ret != 0 && ret != ETIMEDOUT); - /* We got the lock now */ - TAILQ_REMOVE(&pending_requests.requests, &sync_req, next); - pthread_mutex_unlock(&pending_requests.lock); + TAILQ_REMOVE(&pending_requests.requests, &pending_req, next); - if (sync_req.reply_received == 0) { + if (pending_req.reply_received == 0) { RTE_LOG(ERR, EAL, "Fail to recv reply for request %s:%s\n", dst, req->name); rte_errno = ETIMEDOUT; return -1; } - if (sync_req.reply_received == -1) { + if (pending_req.reply_received == -1) { RTE_LOG(DEBUG, EAL, "Asked to ignore response\n"); /* not receiving this message is not an error, so decrement * number of sent messages @@ -956,56 +960,69 @@ mp_request_sync(const char *dst, struct rte_mp_msg *req, return 0; } -int __rte_experimental +int rte_mp_request_sync(struct rte_mp_msg *req, struct rte_mp_reply *reply, const struct timespec *ts) { - int dir_fd, ret = 0; + int dir_fd, ret = -1; DIR *mp_dir; struct dirent *ent; struct timeval now; struct timespec end; + const struct internal_config *internal_conf = + eal_get_internal_configuration(); RTE_LOG(DEBUG, EAL, "request: %s\n", req->name); - if (check_input(req) == false) + reply->nb_sent = 0; + reply->nb_received = 0; + reply->msgs = NULL; + + if (check_input(req) != 0) + goto end; + + if (internal_conf->no_shconf) { + RTE_LOG(DEBUG, EAL, "No shared files mode enabled, IPC is disabled\n"); + rte_errno = ENOTSUP; return -1; + } + if (gettimeofday(&now, NULL) < 0) { - RTE_LOG(ERR, EAL, "Faile to get current time\n"); + RTE_LOG(ERR, EAL, "Failed to get current time\n"); rte_errno = errno; - return -1; + goto end; } end.tv_nsec = (now.tv_usec * 1000 + ts->tv_nsec) % 1000000000; end.tv_sec = now.tv_sec + ts->tv_sec + (now.tv_usec * 1000 + ts->tv_nsec) / 1000000000; - reply->nb_sent = 0; - reply->nb_received = 0; - reply->msgs = NULL; - /* for secondary process, send request to the primary process only */ - if (rte_eal_process_type() == RTE_PROC_SECONDARY) - return mp_request_sync(eal_mp_socket_path(), req, reply, &end); + if (rte_eal_process_type() == RTE_PROC_SECONDARY) { + pthread_mutex_lock(&pending_requests.lock); + ret = mp_request_sync(eal_mp_socket_path(), req, reply, &end); + pthread_mutex_unlock(&pending_requests.lock); + goto end; + } /* for primary process, broadcast request, and collect reply 1 by 1 */ mp_dir = opendir(mp_dir_path); if (!mp_dir) { RTE_LOG(ERR, EAL, "Unable to open directory %s\n", mp_dir_path); rte_errno = errno; - return -1; + goto end; } dir_fd = dirfd(mp_dir); /* lock the directory to prevent processes spinning up while we send */ - if (flock(dir_fd, LOCK_EX)) { + if (flock(dir_fd, LOCK_SH)) { RTE_LOG(ERR, EAL, "Unable to lock directory %s\n", mp_dir_path); - closedir(mp_dir); rte_errno = errno; - return -1; + goto close_end; } + pthread_mutex_lock(&pending_requests.lock); while ((ent = readdir(mp_dir))) { char path[PATH_MAX]; @@ -1015,18 +1032,33 @@ rte_mp_request_sync(struct rte_mp_msg *req, struct rte_mp_reply *reply, snprintf(path, sizeof(path), "%s/%s", mp_dir_path, ent->d_name); + /* unlocks the mutex while waiting for response, + * locks on receive + */ if (mp_request_sync(path, req, reply, &end)) - ret = -1; + goto unlock_end; } + ret = 0; + +unlock_end: + pthread_mutex_unlock(&pending_requests.lock); /* unlock the directory */ flock(dir_fd, LOCK_UN); +close_end: /* dir_fd automatically closed on closedir */ closedir(mp_dir); + +end: + if (ret) { + free(reply->msgs); + reply->nb_received = 0; + reply->msgs = NULL; + } return ret; } -int __rte_experimental +int rte_mp_request_async(struct rte_mp_msg *req, const struct timespec *ts, rte_mp_async_reply_t clb) { @@ -1040,29 +1072,34 @@ rte_mp_request_async(struct rte_mp_msg *req, const struct timespec *ts, struct timeval now; struct timespec *end; bool dummy_used = false; + const struct internal_config *internal_conf = + eal_get_internal_configuration(); RTE_LOG(DEBUG, EAL, "request: %s\n", req->name); - if (check_input(req) == false) + if (check_input(req) != 0) + return -1; + + if (internal_conf->no_shconf) { + RTE_LOG(DEBUG, EAL, "No shared files mode enabled, IPC is disabled\n"); + rte_errno = ENOTSUP; return -1; + } + if (gettimeofday(&now, NULL) < 0) { - RTE_LOG(ERR, EAL, "Faile to get current time\n"); + RTE_LOG(ERR, EAL, "Failed to get current time\n"); rte_errno = errno; return -1; } - copy = malloc(sizeof(*copy)); - dummy = malloc(sizeof(*dummy)); - param = malloc(sizeof(*param)); + copy = calloc(1, sizeof(*copy)); + dummy = calloc(1, sizeof(*dummy)); + param = calloc(1, sizeof(*param)); if (copy == NULL || dummy == NULL || param == NULL) { RTE_LOG(ERR, EAL, "Failed to allocate memory for async reply\n"); rte_errno = ENOMEM; goto fail; } - memset(copy, 0, sizeof(*copy)); - memset(dummy, 0, sizeof(*dummy)); - memset(param, 0, sizeof(*param)); - /* copy message */ memcpy(copy, req, sizeof(*copy)); @@ -1096,7 +1133,7 @@ rte_mp_request_async(struct rte_mp_msg *req, const struct timespec *ts, /* for secondary process, send request to the primary process only */ if (rte_eal_process_type() == RTE_PROC_SECONDARY) { - ret = mp_request_async(eal_mp_socket_path(), copy, param); + ret = mp_request_async(eal_mp_socket_path(), copy, param, ts); /* if we didn't send anything, put dummy request on the queue */ if (ret == 0 && reply->nb_sent == 0) { @@ -1123,7 +1160,7 @@ rte_mp_request_async(struct rte_mp_msg *req, const struct timespec *ts, dir_fd = dirfd(mp_dir); /* lock the directory to prevent processes spinning up while we send */ - if (flock(dir_fd, LOCK_EX)) { + if (flock(dir_fd, LOCK_SH)) { RTE_LOG(ERR, EAL, "Unable to lock directory %s\n", mp_dir_path); rte_errno = errno; @@ -1139,7 +1176,7 @@ rte_mp_request_async(struct rte_mp_msg *req, const struct timespec *ts, snprintf(path, sizeof(path), "%s/%s", mp_dir_path, ent->d_name); - if (mp_request_async(path, copy, param)) + if (mp_request_async(path, copy, param, ts)) ret = -1; } /* if we didn't send anything, put dummy request on the queue */ @@ -1148,9 +1185,6 @@ rte_mp_request_async(struct rte_mp_msg *req, const struct timespec *ts, dummy_used = true; } - /* trigger async request thread wake up */ - pthread_cond_signal(&pending_requests.async_cond); - /* finally, unlock the queue */ pthread_mutex_unlock(&pending_requests.lock); @@ -1176,12 +1210,14 @@ fail: return -1; } -int __rte_experimental +int rte_mp_reply(struct rte_mp_msg *msg, const char *peer) { RTE_LOG(DEBUG, EAL, "reply: %s\n", msg->name); + const struct internal_config *internal_conf = + eal_get_internal_configuration(); - if (check_input(msg) == false) + if (check_input(msg) != 0) return -1; if (peer == NULL) { @@ -1190,5 +1226,50 @@ rte_mp_reply(struct rte_mp_msg *msg, const char *peer) return -1; } + if (internal_conf->no_shconf) { + RTE_LOG(DEBUG, EAL, "No shared files mode enabled, IPC is disabled\n"); + return 0; + } + return mp_send(msg, peer, MP_REP); } + +/* Internally, the status of the mp feature is represented as a three-state: + * - "unknown" as long as no secondary process attached to a primary process + * and there was no call to rte_mp_disable yet, + * - "enabled" as soon as a secondary process attaches to a primary process, + * - "disabled" when a primary process successfully called rte_mp_disable, + */ +enum mp_status { + MP_STATUS_UNKNOWN, + MP_STATUS_DISABLED, + MP_STATUS_ENABLED, +}; + +static bool +set_mp_status(enum mp_status status) +{ + struct rte_mem_config *mcfg = rte_eal_get_configuration()->mem_config; + uint8_t expected; + uint8_t desired; + + expected = MP_STATUS_UNKNOWN; + desired = status; + if (__atomic_compare_exchange_n(&mcfg->mp_status, &expected, desired, + false, __ATOMIC_RELAXED, __ATOMIC_RELAXED)) + return true; + + return __atomic_load_n(&mcfg->mp_status, __ATOMIC_RELAXED) == desired; +} + +bool +rte_mp_disable(void) +{ + return set_mp_status(MP_STATUS_DISABLED); +} + +bool +__rte_mp_enable(void) +{ + return set_mp_status(MP_STATUS_ENABLED); +}