#include <sys/un.h>
#include <unistd.h>
+#include <rte_alarm.h>
#include <rte_common.h>
#include <rte_cycles.h>
#include <rte_eal.h>
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 */
};
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;
if (strlen(name) > 0)
snprintf(buf, len, "%s_%s", prefix, name);
else
- snprintf(buf, len, "%s", prefix);
+ strlcpy(buf, prefix, len);
}
int
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);
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;
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;
}
* 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",
}
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 */
};
{
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;
/* 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;
} 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;
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);
-
- /* we exit this function holding the lock */
- wait_for_async_messages();
-
- if (gettimeofday(&now, NULL) < 0) {
- 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);
- }
+ 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
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");
+ return 0;
+ }
/* 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);
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) {
+ 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);
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);
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));
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;
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;
+ struct pending_request *pending_req, *exist;
int ret;
- 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);
+ 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 = 0;
goto fail;
}
- TAILQ_INSERT_TAIL(&pending_requests.requests, sync_req, next);
+ TAILQ_INSERT_TAIL(&pending_requests.requests, pending_req, next);
param->user_reply.nb_sent++;
+ 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);
+ rte_panic("Fix the above shit to properly free all memory\n");
+ }
+
return 0;
fail:
- free(sync_req);
+ free(pending_req);
free(reply_msg);
return ret;
}
{
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);
+ 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;
}
} else if (ret == 0)
return 0;
- TAILQ_INSERT_TAIL(&pending_requests.requests, &sync_req, next);
+ 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
if (check_input(req) == false)
return -1;
+
+ if (internal_config.no_shconf) {
+ RTE_LOG(DEBUG, EAL, "No shared files mode enabled, IPC is disabled\n");
+ return 0;
+ }
+
if (gettimeofday(&now, NULL) < 0) {
RTE_LOG(ERR, EAL, "Faile to get current time\n");
rte_errno = errno;
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);
+ return ret;
+ }
/* for primary process, broadcast request, and collect reply 1 by 1 */
mp_dir = opendir(mp_dir_path);
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);
return -1;
}
+ pthread_mutex_lock(&pending_requests.lock);
while ((ent = readdir(mp_dir))) {
char path[PATH_MAX];
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;
}
+ pthread_mutex_unlock(&pending_requests.lock);
/* unlock the directory */
flock(dir_fd, LOCK_UN);
if (check_input(req) == false)
return -1;
+
+ if (internal_config.no_shconf) {
+ RTE_LOG(DEBUG, EAL, "No shared files mode enabled, IPC is disabled\n");
+ return 0;
+ }
+
if (gettimeofday(&now, NULL) < 0) {
RTE_LOG(ERR, EAL, "Faile 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));
/* 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) {
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;
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 */
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);
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);
}
git.droids-corp.org / dpdk.git / blobdiff