X-Git-Url: http://git.droids-corp.org/?a=blobdiff_plain;f=lib%2Flibrte_eal%2Fcommon%2Feal_common_proc.c;h=935e8fefeba85800e561b70f0ee9865dcc00fb73;hb=b86aebcb6f22a3f282eaa5624764f827988b48b9;hp=b704f5af8c24403dc681e90b2845314442c5146e;hpb=ce3a7312357ba636dea4e326e5676615502805c5;p=dpdk.git diff --git a/lib/librte_eal/common/eal_common_proc.c b/lib/librte_eal/common/eal_common_proc.c index b704f5af8c..935e8fefeb 100644 --- a/lib/librte_eal/common/eal_common_proc.c +++ b/lib/librte_eal/common/eal_common_proc.c @@ -20,12 +20,14 @@ #include #include +#include #include #include #include #include #include #include +#include #include "eal_private.h" #include "eal_filesystem.h" @@ -35,6 +37,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; @@ -60,13 +63,32 @@ struct mp_msg_internal { struct rte_mp_msg msg; }; +struct async_request_param { + rte_mp_async_reply_t clb; + struct rte_mp_reply user_reply; + struct timespec end; + int n_responses_processed; +}; + struct pending_request { TAILQ_ENTRY(pending_request) next; - int reply_received; + enum { + REQUEST_TYPE_SYNC, + REQUEST_TYPE_ASYNC + } type; char dst[PATH_MAX]; struct rte_mp_msg *request; struct rte_mp_msg *reply; - pthread_cond_t cond; + int reply_received; + RTE_STD_C11 + union { + struct { + struct async_request_param *param; + } async; + struct { + pthread_cond_t cond; + } sync; + }; }; TAILQ_HEAD(pending_request_list, pending_request); @@ -76,16 +98,27 @@ static struct { pthread_mutex_t lock; } pending_requests = { .requests = TAILQ_HEAD_INITIALIZER(pending_requests.requests), - .lock = PTHREAD_MUTEX_INITIALIZER + .lock = PTHREAD_MUTEX_INITIALIZER, + /**< used in async requests only */ }; /* forward declarations */ 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 * -find_sync_request(const char *dst, const char *act_name) +async_reply_handle_thread_unsafe(void *arg); + +static void +trigger_async_action(struct pending_request *req); + +static struct pending_request * +find_pending_request(const char *dst, const char *act_name) { struct pending_request *r; @@ -106,7 +139,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 @@ -164,20 +197,26 @@ 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; - if (validate_action_name(name)) + if (validate_action_name(name) != 0) + return -1; + + if (internal_config.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); @@ -192,13 +231,18 @@ 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; - if (validate_action_name(name)) + if (validate_action_name(name) != 0) + return; + + if (internal_config.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); @@ -239,7 +283,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; } @@ -252,14 +296,22 @@ 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; @@ -267,16 +319,27 @@ process_msg(struct mp_msg_internal *m, struct sockaddr_un *s) 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; - pthread_cond_signal(&sync_req->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; } @@ -294,7 +357,9 @@ process_msg(struct mp_msg_internal *m, struct sockaddr_un *s) * 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", @@ -319,12 +384,156 @@ mp_handle(void *arg __rte_unused) return NULL; } +static int +timespec_cmp(const struct timespec *a, const struct timespec *b) +{ + if (a->tv_sec < b->tv_sec) + return -1; + if (a->tv_sec > b->tv_sec) + return 1; + if (a->tv_nsec < b->tv_nsec) + return -1; + if (a->tv_nsec > b->tv_nsec) + return 1; + return 0; +} + +enum async_action { + ACTION_FREE, /**< free the action entry, but don't trigger callback */ + ACTION_TRIGGER /**< trigger callback, then free action entry */ +}; + +static enum async_action +process_async_request(struct pending_request *sr, const struct timespec *now) +{ + struct async_request_param *param; + struct rte_mp_reply *reply; + bool timeout, last_msg; + + param = sr->async.param; + reply = ¶m->user_reply; + + /* did we timeout? */ + timeout = timespec_cmp(¶m->end, now) <= 0; + + /* if we received a response, adjust relevant data and copy mesasge. */ + if (sr->reply_received == 1 && sr->reply) { + struct rte_mp_msg *msg, *user_msgs, *tmp; + + msg = sr->reply; + user_msgs = reply->msgs; + + tmp = realloc(user_msgs, sizeof(*msg) * + (reply->nb_received + 1)); + if (!tmp) { + RTE_LOG(ERR, EAL, "Fail to alloc reply for request %s:%s\n", + sr->dst, sr->request->name); + /* this entry is going to be removed and its message + * dropped, but we don't want to leak memory, so + * continue. + */ + } else { + user_msgs = tmp; + reply->msgs = user_msgs; + memcpy(&user_msgs[reply->nb_received], + msg, sizeof(*msg)); + reply->nb_received++; + } + + /* mark this request as processed */ + param->n_responses_processed++; + } 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; + + return last_msg ? ACTION_TRIGGER : ACTION_FREE; +} + +static void +trigger_async_action(struct pending_request *sr) +{ + struct async_request_param *param; + struct rte_mp_reply *reply; + + param = sr->async.param; + reply = ¶m->user_reply; + + param->clb(sr->request, reply); + + /* clean up */ + free(sr->async.param->user_reply.msgs); + free(sr->async.param); + free(sr->request); + free(sr); +} + +static struct pending_request * +async_reply_handle_thread_unsafe(void *arg) +{ + struct pending_request *req = (struct pending_request *)arg; + enum async_action action; + struct timespec ts_now; + struct timeval now; + + 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; + + action = process_async_request(req, &ts_now); + + TAILQ_REMOVE(&pending_requests.requests, req, next); + + 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; + } + RTE_LOG(ERR, EAL, "Failed to cancel alarm\n"); + } + + if (action == ACTION_TRIGGER) + return req; +no_trigger: + free(req); + return NULL; +} + +static void +async_reply_handle(void *arg) +{ + struct pending_request *req; + + pthread_mutex_lock(&pending_requests.lock); + req = async_reply_handle_thread_unsafe(arg); + pthread_mutex_unlock(&pending_requests.lock); + + 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()); @@ -353,44 +562,42 @@ 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 tid; + pthread_t mp_handle_tid; + + /* in no shared files mode, we do not have secondary processes support, + * so no need to initialize IPC. + */ + if (internal_config.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); @@ -407,19 +614,13 @@ 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(&tid, NULL, mp_handle, NULL) < 0) { + if (rte_ctrl_thread_create(&mp_handle_tid, "rte_mp_handle", + NULL, mp_handle, NULL) < 0) { RTE_LOG(ERR, EAL, "failed to create mp thead: %s\n", strerror(errno)); close(mp_fd); @@ -428,10 +629,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(tid, thread_name); - /* unlock the directory */ flock(dir_fd, LOCK_UN); close(dir_fd); @@ -439,6 +636,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. @@ -462,7 +665,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)); @@ -495,11 +698,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; @@ -536,7 +734,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; @@ -563,66 +761,143 @@ 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)) - return false; + 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 (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)) + if (check_input(msg) != 0) return -1; + if (internal_config.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_one(const char *dst, struct rte_mp_msg *req, +mp_request_async(const char *dst, struct rte_mp_msg *req, + struct async_request_param *param, const struct timespec *ts) +{ + struct rte_mp_msg *reply_msg; + struct pending_request *pending_req, *exist; + int ret = -1; + + 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; + } + + 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_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; + goto fail; + } + + ret = send_msg(dst, req, MP_REQ); + if (ret < 0) { + RTE_LOG(ERR, EAL, "Fail to send request %s:%s\n", + dst, req->name); + ret = -1; + goto fail; + } else if (ret == 0) { + 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(pending_req); + free(reply_msg); + return ret; +} + +static int +mp_request_sync(const char *dst, struct rte_mp_msg *req, struct rte_mp_reply *reply, const struct timespec *ts) { int ret; struct rte_mp_msg msg, *tmp; - struct pending_request sync_req, *exist; + struct pending_request pending_req, *exist; - sync_req.reply_received = 0; - strcpy(sync_req.dst, dst); - sync_req.request = req; - sync_req.reply = &msg; - pthread_cond_init(&sync_req.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; } @@ -634,24 +909,24 @@ mp_request_one(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.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 @@ -673,11 +948,11 @@ mp_request_one(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; @@ -685,42 +960,195 @@ rte_mp_request_sync(struct rte_mp_msg *req, struct rte_mp_reply *reply, 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_config.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; + /* for secondary process, send request to the primary process only */ + 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; + goto end; + } + + dir_fd = dirfd(mp_dir); + /* lock the directory to prevent processes spinning up while we send */ + if (flock(dir_fd, LOCK_SH)) { + RTE_LOG(ERR, EAL, "Unable to lock directory %s\n", + mp_dir_path); + rte_errno = errno; + goto close_end; + } + + pthread_mutex_lock(&pending_requests.lock); + while ((ent = readdir(mp_dir))) { + char path[PATH_MAX]; + + if (fnmatch(mp_filter, ent->d_name, 0) != 0) + continue; + + 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)) + 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_mp_request_async(struct rte_mp_msg *req, const struct timespec *ts, + rte_mp_async_reply_t clb) +{ + struct rte_mp_msg *copy; + struct pending_request *dummy; + struct async_request_param *param; + struct rte_mp_reply *reply; + int dir_fd, ret = 0; + DIR *mp_dir; + struct dirent *ent; + struct timeval now; + struct timespec *end; + bool dummy_used = false; + + RTE_LOG(DEBUG, EAL, "request: %s\n", req->name); + + if (check_input(req) != 0) + return -1; + + if (internal_config.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, "Failed to get current time\n"); + rte_errno = errno; + return -1; + } + 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; + } + + /* copy message */ + memcpy(copy, req, sizeof(*copy)); + + param->n_responses_processed = 0; + param->clb = clb; + end = ¶m->end; + reply = ¶m->user_reply; + + 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; + /* we have to lock the request queue here, as we will be adding a bunch + * of requests to the queue at once, and some of the replies may arrive + * before we add all of the requests to the queue. + */ + pthread_mutex_lock(&pending_requests.lock); + + /* we have to ensure that callback gets triggered even if we don't send + * anything, therefore earlier we have allocated a dummy request. fill + * it, and put it on the queue if we don't send any requests. + */ + dummy->type = REQUEST_TYPE_ASYNC; + dummy->request = copy; + dummy->reply = NULL; + dummy->async.param = param; + dummy->reply_received = 1; /* short-circuit the timeout */ + /* for secondary process, send request to the primary process only */ - if (rte_eal_process_type() == RTE_PROC_SECONDARY) - return mp_request_one(eal_mp_socket_path(), req, reply, &end); + if (rte_eal_process_type() == RTE_PROC_SECONDARY) { + 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) { + TAILQ_INSERT_TAIL(&pending_requests.requests, dummy, + next); + dummy_used = true; + } - /* for primary process, broadcast request, and collect reply 1 by 1 */ + pthread_mutex_unlock(&pending_requests.lock); + + /* if we couldn't send anything, clean up */ + if (ret != 0) + goto fail; + return 0; + } + + /* for primary process, broadcast request */ 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 unlock_fail; } - 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 closedir_fail; } while ((ent = readdir(mp_dir))) { @@ -732,24 +1160,46 @@ 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); - if (mp_request_one(path, req, reply, &end)) + if (mp_request_async(path, copy, param, ts)) ret = -1; } + /* if we didn't send anything, put dummy request on the queue */ + if (ret == 0 && reply->nb_sent == 0) { + TAILQ_INSERT_HEAD(&pending_requests.requests, dummy, next); + dummy_used = true; + } + + /* finally, unlock the queue */ + pthread_mutex_unlock(&pending_requests.lock); + /* unlock the directory */ flock(dir_fd, LOCK_UN); /* dir_fd automatically closed on closedir */ closedir(mp_dir); + + /* if dummy was unused, free it */ + if (!dummy_used) + free(dummy); + return ret; +closedir_fail: + closedir(mp_dir); +unlock_fail: + pthread_mutex_unlock(&pending_requests.lock); +fail: + free(dummy); + free(param); + free(copy); + 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); - if (check_input(msg) == false) + if (check_input(msg) != 0) return -1; if (peer == NULL) { @@ -758,5 +1208,10 @@ rte_mp_reply(struct rte_mp_msg *msg, const char *peer) return -1; } + if (internal_config.no_shconf) { + RTE_LOG(DEBUG, EAL, "No shared files mode enabled, IPC is disabled\n"); + return 0; + } + return mp_send(msg, peer, MP_REP); }