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;
if (m->type == MP_REP || m->type == MP_IGN) {
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;
+ pending_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)
+ 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);
} else
* yet ready to process this request.
*/
struct rte_mp_msg dummy;
+
memset(&dummy, 0, sizeof(dummy));
- snprintf(dummy.name, sizeof(dummy.name),
- "%s", msg->name);
+ 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",
} 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;
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;
}
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;
/* 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) {
+ 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);
+ close(dir_fd);
mp_fd = -1;
return -1;
}
- if (pthread_create(&async_reply_handle_tid, NULL,
+ if (rte_ctrl_thread_create(&async_reply_handle_tid,
+ "rte_mp_async", NULL,
async_reply_handle, NULL) < 0) {
RTE_LOG(ERR, EAL, "failed to create mp thead: %s\n",
strerror(errno));
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));
struct async_request_param *param)
{
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));
+ strcpy(pending_req->dst, 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++;
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
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);
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);
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));
git.droids-corp.org / dpdk.git / blobdiff