X-Git-Url: http://git.droids-corp.org/?a=blobdiff_plain;f=lib%2Flibrte_eal%2Fcommon%2Feal_common_proc.c;h=cbe8d10fcfee9dc034b716ef62582cd1453c5ae9;hb=85458c2f3e7b525d14695ed78ce3a250cb85837e;hp=c450c848f42bfb6e3cb6cf01b3c11e1275619f31;hpb=dec7b1884a555c941110526cc6cb9dfa08fd51b9;p=dpdk.git diff --git a/lib/librte_eal/common/eal_common_proc.c b/lib/librte_eal/common/eal_common_proc.c index c450c848f4..cbe8d10fcf 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 @@ -36,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; @@ -94,11 +96,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,6 +106,16 @@ 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_pending_request(const char *dst, const char *act_name) @@ -187,14 +197,20 @@ 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; @@ -215,14 +231,19 @@ 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); if (entry == NULL) { @@ -275,7 +296,15 @@ 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; } @@ -290,6 +319,8 @@ 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); pending_req = find_pending_request(s->sun_path, msg->name); if (pending_req) { @@ -301,11 +332,14 @@ process_msg(struct mp_msg_internal *m, struct sockaddr_un *s) if (pending_req->type == REQUEST_TYPE_SYNC) pthread_cond_signal(&pending_req->sync.cond); else if (pending_req->type == REQUEST_TYPE_ASYNC) - pthread_cond_signal( - &pending_requests.async_cond); + 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; } @@ -365,7 +399,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 */ }; @@ -375,7 +408,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; @@ -383,13 +416,6 @@ 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 (sr->reply_received == 1 && sr->reply) { struct rte_mp_msg *msg, *user_msgs, *tmp; @@ -448,126 +474,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 struct pending_request * -check_trigger(struct timespec *ts) +async_reply_handle_thread_unsafe(void *arg) { - struct pending_request *next, *cur, *trigger = NULL; - - TAILQ_FOREACH_SAFE(cur, &pending_requests.requests, next, next) { - enum async_action action; - if (cur->type != REQUEST_TYPE_ASYNC) - continue; + struct pending_request *req = (struct pending_request *)arg; + enum async_action action; + struct timespec ts_now; + struct timeval now; - action = process_async_request(cur, ts); - if (action == ACTION_FREE) { - TAILQ_REMOVE(&pending_requests.requests, cur, next); - free(cur); - } else if (action == ACTION_TRIGGER) { - TAILQ_REMOVE(&pending_requests.requests, cur, next); - trigger = cur; - break; - } + if (gettimeofday(&now, NULL) < 0) { + RTE_LOG(ERR, EAL, "Cannot get current time\n"); + goto no_trigger; } - return trigger; -} + ts_now.tv_nsec = now.tv_usec * 1000; + ts_now.tv_sec = now.tv_sec; -static void -wait_for_async_messages(void) -{ - struct pending_request *sr; - struct timespec timeout; - bool timedwait = false; - bool nowait = false; - int ret; + action = process_async_request(req, &ts_now); - /* 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; - } + TAILQ_REMOVE(&pending_requests.requests, req, next); - /* sometimes, we don't even wait */ - if (sr->reply_received) { - nowait = true; - 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; } + RTE_LOG(ERR, EAL, "Failed to cancel alarm\n"); } - if (nowait) - return; - - do { - ret = timedwait ? - pthread_cond_timedwait( - &pending_requests.async_cond, - &pending_requests.lock, - &timeout) : - pthread_cond_wait( - &pending_requests.async_cond, - &pending_requests.lock); - } while (ret != 0 && ret != ETIMEDOUT); - - /* we've been woken up or timed out */ + if (action == ACTION_TRIGGER) + return req; +no_trigger: + free(req); + return NULL; } -static void * -async_reply_handle(void *arg __rte_unused) +static void +async_reply_handle(void *arg) { - struct timeval now; - struct timespec ts_now; - while (1) { - struct pending_request *trigger = NULL; - - pthread_mutex_lock(&pending_requests.lock); + struct pending_request *req; - /* we exit this function holding the lock */ - wait_for_async_messages(); - - if (gettimeofday(&now, NULL) < 0) { - pthread_mutex_unlock(&pending_requests.lock); - RTE_LOG(ERR, EAL, "Cannot get current time\n"); - break; - } - ts_now.tv_nsec = now.tv_usec * 1000; - ts_now.tv_sec = now.tv_sec; - - do { - trigger = check_trigger(&ts_now); - /* unlock request list */ - pthread_mutex_unlock(&pending_requests.lock); - - if (trigger) { - trigger_async_action(trigger); - free(trigger); - - /* we've triggered a callback, but there may be - * more, so lock the list and check again. - */ - pthread_mutex_lock(&pending_requests.lock); - } - } while (trigger); - } - - 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()); @@ -596,36 +562,34 @@ 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; + + /* 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)); @@ -650,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(&mp_handle_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); @@ -671,24 +629,6 @@ rte_mp_channel_init(void) 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", - strerror(errno)); - close(mp_fd); - close(dir_fd); - mp_fd = -1; - return -1; - } - - /* try best to set thread name */ - strlcpy(thread_name, "rte_mp_handle", sizeof(thread_name)); - rte_thread_setname(mp_handle_tid, thread_name); - - /* try best to set thread name */ - strlcpy(thread_name, "rte_mp_async_handle", sizeof(thread_name)); - rte_thread_setname(async_reply_handle_tid, thread_name); - /* unlock the directory */ flock(dir_fd, LOCK_UN); close(dir_fd); @@ -696,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. @@ -752,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; @@ -793,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; @@ -820,51 +761,69 @@ 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)) + 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_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 *pending_req, *exist; - int ret; + int ret = -1; pending_req = calloc(1, sizeof(*pending_req)); reply_msg = calloc(1, sizeof(*reply_msg)); @@ -877,7 +836,6 @@ mp_request_async(const char *dst, struct rte_mp_msg *req, pending_req->type = REQUEST_TYPE_ASYNC; strlcpy(pending_req->dst, dst, sizeof(pending_req->dst)); - strcpy(pending_req->dst, dst); pending_req->request = req; pending_req->reply = reply_msg; pending_req->async.param = param; @@ -902,10 +860,18 @@ mp_request_async(const char *dst, struct rte_mp_msg *req, ret = 0; goto fail; } - TAILQ_INSERT_TAIL(&pending_requests.requests, pending_req, next); - 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); @@ -982,11 +948,11 @@ 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; @@ -994,28 +960,35 @@ 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; - 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) { pthread_mutex_lock(&pending_requests.lock); ret = mp_request_sync(eal_mp_socket_path(), req, reply, &end); pthread_mutex_unlock(&pending_requests.lock); - return ret; + goto end; } /* for primary process, broadcast request, and collect reply 1 by 1 */ @@ -1023,17 +996,16 @@ rte_mp_request_sync(struct rte_mp_msg *req, struct rte_mp_reply *reply, 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); @@ -1050,18 +1022,29 @@ rte_mp_request_sync(struct rte_mp_msg *req, struct rte_mp_reply *reply, * 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) { @@ -1078,8 +1061,15 @@ rte_mp_request_async(struct rte_mp_msg *req, const struct timespec *ts, RTE_LOG(DEBUG, EAL, "request: %s\n", req->name); - if (check_input(req) == false) + 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, "Faile to get current time\n"); rte_errno = errno; @@ -1127,7 +1117,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) { @@ -1154,7 +1144,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; @@ -1170,7 +1160,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 */ @@ -1179,9 +1169,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); @@ -1207,12 +1194,12 @@ 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); - if (check_input(msg) == false) + if (check_input(msg) != 0) return -1; if (peer == NULL) { @@ -1221,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); }