xuzhenbao commented on code in PR #437: URL: https://github.com/apache/celix/pull/437#discussion_r956092902
########## bundles/remote_services/remote_service_admin_shm_v2/rsa_shm/src/rsa_shm_server.c: ########## @@ -0,0 +1,316 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +#include <rsa_shm_server.h> +#include <rsa_shm_msg.h> +#include <rsa_shm_constants.h> +#include <shm_cache.h> +#include <celix_log_helper.h> +#include <celix_api.h> +#include <thpool.h> +#include <sys/un.h> +#include <sys/socket.h> +#include <sys/types.h> +#include <stdlib.h> +#include <unistd.h> +#include <sys/uio.h> +#include <sys/param.h> +#include <assert.h> +#include <stdbool.h> +#include <errno.h> + +#define MAX_RSA_SHM_SERVER_HANDLE_MSG_THREADS_NUM 5 + +struct rsa_shm_server { + celix_bundle_context_t *ctx; + char *name; + celix_log_helper_t *loghelper; + int sfd; + shm_cache_t *shmCache; + threadpool threadPool; + celix_thread_t revMsgThread; + bool revMsgThreadActive; + rsaShmServer_receiveMsgCB revCB; + void *revCBHandle; + long msgTimeOutInSec; +}; + +struct rsa_shm_server_thpool_work_data { + rsa_shm_server_t *server; + rsa_shm_msg_control_t *msgCtrl; + void *msgBuffer; + size_t maxBufferSize; + size_t metadataSize; + size_t requestSize; +}; + +static void *rsaShmServer_receiveMsgThread(void *data); + +celix_status_t rsaShmServer_create(celix_bundle_context_t *ctx, const char *name, celix_log_helper_t *loghelper, + rsaShmServer_receiveMsgCB receiveCB, void *revHandle, rsa_shm_server_t **shmServerOut) { + int status = CELIX_SUCCESS; + if (name == NULL || ctx == NULL || strlen(name) >= MAX_RSA_SHM_SERVER_NAME_SIZE + || loghelper == NULL || receiveCB == NULL) { + return CELIX_ILLEGAL_ARGUMENT; + } + + rsa_shm_server_t *server = (rsa_shm_server_t *)calloc(1, sizeof(rsa_shm_server_t)); + assert(server != NULL); + server->ctx = ctx; + server->msgTimeOutInSec = celix_bundleContext_getPropertyAsLong(ctx, + RSA_SHM_MSG_TIMEOUT_KEY, RSA_SHM_MSG_TIMEOUT_DEFAULT_IN_S); + server->name = strdup(name); + assert(server->name != NULL); + server->loghelper = loghelper; + int sfd = socket(AF_UNIX, SOCK_DGRAM, 0); + if (sfd == -1) { + celix_logHelper_error(loghelper, "RsaShmServer: create socket fd err, errno is %d.", errno); + status = CELIX_ERROR_MAKE(CELIX_FACILITY_CERRNO, errno); + goto sfd_err; + } + server->sfd = sfd; + struct sockaddr_un svaddr; + memset(&svaddr, 0, sizeof(svaddr)); + svaddr.sun_family = AF_UNIX; + strncpy(&svaddr.sun_path[1], name, sizeof(svaddr.sun_path) - 1); + if (bind(sfd, (struct sockaddr *) &svaddr, sizeof(struct sockaddr_un)) == -1) { + celix_logHelper_error(loghelper, "RsaShmServer: bind socket fd err, errno is %d.", errno); + status = CELIX_ERROR_MAKE(CELIX_FACILITY_CERRNO, errno); + goto sfd_bind_err; + } + + shm_cache_t *shmCache = NULL; + status = shmCache_create(false, &shmCache); + if (status != CELIX_SUCCESS) { + celix_logHelper_error(loghelper, "RsaShmServer: create shm cache err; error code is %d.", status); + goto create_shm_cache_err; + } + server->shmCache = shmCache; + + server->threadPool = thpool_init(MAX_RSA_SHM_SERVER_HANDLE_MSG_THREADS_NUM); + if (server->threadPool == NULL) { + celix_logHelper_error(loghelper, "RsaShmServer: create thread pool err."); + status = CELIX_ILLEGAL_STATE; + goto create_thpool_err; + } + server->revCB = receiveCB; + server->revCBHandle = revHandle; + server->revMsgThreadActive = true; + status = celixThread_create(&server->revMsgThread, NULL, rsaShmServer_receiveMsgThread, server); + if (status != CELIX_SUCCESS) { + celix_logHelper_error(loghelper, "RsaShmServer: create receive msg thread err."); + goto create_rev_msg_thread_err; + } + *shmServerOut = server; + return CELIX_SUCCESS; +create_rev_msg_thread_err: + thpool_destroy(server->threadPool); +create_thpool_err: + shmCache_destroy(shmCache); +create_shm_cache_err: +sfd_bind_err: + close(sfd); +sfd_err: + free(server->name); + free(server); + return status; +} + +void rsaShmServer_destroy(rsa_shm_server_t *server) { + if (server == NULL) { + return; + } + server->revMsgThreadActive = false; + shutdown(server->sfd,SHUT_RD); + celixThread_join(server->revMsgThread, NULL); + thpool_destroy(server->threadPool); + shmCache_destroy(server->shmCache); + close(server->sfd); + free(server->name); + free(server); + return; +} + + +static void rsaShmServer_terminateMsgHandling(rsa_shm_msg_control_t *ctrl) { + assert(ctrl != NULL); + + // weakup client, terminate current interaction + pthread_mutex_lock(&ctrl->lock); + ctrl->msgState = ABEND; + //Signaling the condition variable first, and then unlocking the mutex, because client will free ctrl when msgState is ABEND. + pthread_cond_signal(&ctrl->signal); Review Comment: This signal is used to wake up `pthread_cond_timedwait`, and the condition variable has the `PTHREAD_PROCESS_SHARED` attribute. The usage like this: `bundles/remote_services/doc/rsa_shm_v2.adoc(IPC for shared memory)`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: dev-unsubscr...@celix.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
