Rebase up to 88ded11f59
Lixia Chen
5 years ago
parent
88ded11f59
commit
983827ec5c
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,163 @@
|
||||
/**
|
||||
* Copyright 2020 Huawei Technologies Co., Ltd
|
||||
|
||||
* Licensed 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 "minddata/dataset/engine/cache/cache_ipc.h"
|
||||
#include <sys/stat.h>
|
||||
|
||||
namespace mindspore {
|
||||
namespace dataset {
|
||||
Status PortToFtok(int port, SharedMemory::shm_key_t *out) {
|
||||
RETURN_UNEXPECTED_IF_NULL(out);
|
||||
key_t shmkey = -1;
|
||||
const std::string unix_path = PortToUnixSocketPath(port);
|
||||
shmkey = ftok(unix_path.data(), 'a');
|
||||
if (shmkey == (key_t)-1) {
|
||||
std::string errMsg = "Unable to create a ftok token. Errno = " + std::to_string(errno);
|
||||
return Status(errno == ENOENT ? StatusCode::kFileNotExist : StatusCode::kUnexpectedError, errMsg);
|
||||
}
|
||||
*out = shmkey;
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
SharedMessage::~SharedMessage() {
|
||||
// Only remove the queue if we are asked to.
|
||||
if (remove_ipc_on_exit_ && msg_qid_ != -1) {
|
||||
// Remove the message que and never mind about the return code.
|
||||
(void)msgctl(msg_qid_, IPC_RMID, nullptr);
|
||||
msg_qid_ = -1;
|
||||
}
|
||||
}
|
||||
|
||||
Status SharedMessage::Create() {
|
||||
CHECK_FAIL_RETURN_UNEXPECTED(msg_qid_ == -1, "Message queue already created");
|
||||
auto access_mode = S_IRUSR | S_IWUSR | S_IROTH | S_IWOTH | S_IRGRP | S_IWGRP;
|
||||
msg_qid_ = msgget(IPC_PRIVATE, IPC_CREAT | IPC_EXCL | access_mode);
|
||||
if (msg_qid_ == -1) {
|
||||
std::string errMsg = "Unable to create a message queue. Errno = " + std::to_string(errno);
|
||||
RETURN_STATUS_UNEXPECTED(errMsg);
|
||||
}
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
Status SharedMessage::SendStatus(const Status &rc) {
|
||||
CHECK_FAIL_RETURN_UNEXPECTED(msg_qid_ != -1, "Invalid message queue id");
|
||||
StatusMsgBuf msg{
|
||||
1,
|
||||
};
|
||||
msg.body.status.err_code = static_cast<int32_t>(rc.get_code());
|
||||
auto err = memcpy_s(msg.body.status.err_msg, kSharedMessageSize, rc.ToString().data(), rc.ToString().size());
|
||||
CHECK_FAIL_RETURN_UNEXPECTED(err == EOK, "memcpy_s failed. err = " + std::to_string(err));
|
||||
msg.body.status.err_msg[rc.ToString().size()] = '\0';
|
||||
err = msgsnd(msg_qid_, reinterpret_cast<void *>(&msg), sizeof(msg.body.status), IPC_NOWAIT);
|
||||
if (err == -1) {
|
||||
std::string errMsg = "Failed to call msgsnd. Errno = " + std::to_string(errno);
|
||||
RETURN_STATUS_UNEXPECTED(errMsg);
|
||||
}
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
Status SharedMessage::ReceiveStatus(Status *rc) {
|
||||
RETURN_UNEXPECTED_IF_NULL(rc);
|
||||
CHECK_FAIL_RETURN_UNEXPECTED(msg_qid_ != -1, "Invalid message queue id");
|
||||
struct StatusMsgBuf msg {};
|
||||
auto err = msgrcv(msg_qid_, reinterpret_cast<void *>(&msg), sizeof(msg.body.status), 0, MSG_NOERROR);
|
||||
if (err == -1) {
|
||||
std::string errMsg = "Failed to call msgrcv. Errno = " + std::to_string(errno);
|
||||
RETURN_STATUS_UNEXPECTED(errMsg);
|
||||
}
|
||||
|
||||
Status rc_recv(static_cast<StatusCode>(msg.body.status.err_code), msg.body.status.err_msg);
|
||||
*rc = std::move(rc_recv);
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
SharedMemory::~SharedMemory() {
|
||||
if (shmat_addr_) {
|
||||
(void)Detach();
|
||||
}
|
||||
if (remove_ipc_on_exit_ && shm_id_ != -1) {
|
||||
// Remove the shared memory and never mind about the return code.
|
||||
Status rc = Destroy();
|
||||
if (rc.IsError()) {
|
||||
MS_LOG(ERROR) << rc.ToString();
|
||||
}
|
||||
}
|
||||
shm_id_ = -1;
|
||||
shmat_addr_ = nullptr;
|
||||
}
|
||||
|
||||
Status SharedMemory::Create(int64_t sz) {
|
||||
auto access_mode = S_IRUSR | S_IWUSR | S_IROTH | S_IWOTH | S_IRGRP | S_IWGRP;
|
||||
shm_id_ = shmget(shm_key_, sz, IPC_CREAT | IPC_EXCL | access_mode);
|
||||
if (shm_id_ == -1) {
|
||||
RETURN_STATUS_UNEXPECTED("Shared memory creation failed. Errno " + std::to_string(errno));
|
||||
} else {
|
||||
shmat_addr_ = shmat(shm_id_, nullptr, 0);
|
||||
if (shmat_addr_ == reinterpret_cast<void *>(-1)) {
|
||||
RETURN_STATUS_UNEXPECTED("Shared memory attach failed. Errno " + std::to_string(errno));
|
||||
}
|
||||
}
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
Status SharedMemory::Attach() {
|
||||
shm_id_ = shmget(shm_key_, 0, 0);
|
||||
if (shm_id_ == -1) {
|
||||
RETURN_STATUS_UNEXPECTED("Shmget failed. Errno " + std::to_string(errno));
|
||||
}
|
||||
shmat_addr_ = shmat(shm_id_, nullptr, 0);
|
||||
if (shmat_addr_ == reinterpret_cast<void *>(-1)) {
|
||||
RETURN_STATUS_UNEXPECTED("Shared memory attach failed. Errno " + std::to_string(errno));
|
||||
}
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
Status SharedMemory::Detach() {
|
||||
if (shmat_addr_) {
|
||||
auto err = shmdt(shmat_addr_);
|
||||
if (err == -1) {
|
||||
RETURN_STATUS_UNEXPECTED("Shared memory detach failed. Errno " + std::to_string(errno));
|
||||
}
|
||||
}
|
||||
shmat_addr_ = nullptr;
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
Status SharedMemory::Destroy() {
|
||||
// Remove the shared memory and never mind about the return code.
|
||||
auto err = shmctl(shm_id_, IPC_RMID, nullptr);
|
||||
if (err == -1) {
|
||||
std::string errMsg = "Unable to remove shared memory with id " + std::to_string(shm_id_);
|
||||
errMsg += ". Errno :" + std::to_string(errno);
|
||||
errMsg += "\nPlesae remove it manually using ipcrm -m command";
|
||||
RETURN_STATUS_UNEXPECTED(errMsg);
|
||||
}
|
||||
return Status::OK();
|
||||
}
|
||||
|
||||
Status SharedMemory::GetNumAttached(int32_t *num) {
|
||||
RETURN_UNEXPECTED_IF_NULL(num);
|
||||
struct shmid_ds ds {};
|
||||
auto err = shmctl(shm_id_, IPC_STAT, &ds);
|
||||
if (err == -1) {
|
||||
std::string errMsg = "Unable to query shared memory with id " + std::to_string(shm_id_);
|
||||
errMsg += "\nPlease remove it manually using ipcrm -m command";
|
||||
RETURN_STATUS_UNEXPECTED(errMsg);
|
||||
}
|
||||
*num = ds.shm_nattch;
|
||||
return Status::OK();
|
||||
}
|
||||
} // namespace dataset
|
||||
} // namespace mindspore
|
||||
@ -0,0 +1,207 @@
|
||||
/**
|
||||
* Copyright 2020 Huawei Technologies Co., Ltd
|
||||
|
||||
* Licensed 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.
|
||||
*/
|
||||
#ifndef MINDSPORE_CCSRC_MINDDATA_DATASET_ENGINE_CACHE_IPC_H_
|
||||
#define MINDSPORE_CCSRC_MINDDATA_DATASET_ENGINE_CACHE_IPC_H_
|
||||
|
||||
#include <sys/types.h>
|
||||
#include <sys/ipc.h>
|
||||
#include <sys/shm.h>
|
||||
#include <sys/msg.h>
|
||||
#include <string>
|
||||
#include <utility>
|
||||
#include "minddata/dataset/engine/cache/cache_common.h"
|
||||
#include "minddata/dataset/util/status.h"
|
||||
|
||||
namespace mindspore {
|
||||
namespace dataset {
|
||||
/// A message queue structure between the parent and the child process
|
||||
struct StatusMsgBuf {
|
||||
int64_t mtype;
|
||||
union {
|
||||
char mtext[1];
|
||||
struct {
|
||||
int32_t err_code;
|
||||
char err_msg[kSharedMessageSize];
|
||||
} status;
|
||||
} body;
|
||||
};
|
||||
|
||||
class BaseIPC {
|
||||
public:
|
||||
BaseIPC() : remove_ipc_on_exit_(false) {}
|
||||
virtual ~BaseIPC() {}
|
||||
/// Indicate if we should remove the ipc resource on exit. Usually this is done by parent process.
|
||||
void RemoveResourcesOnExit() { remove_ipc_on_exit_ = true; }
|
||||
/// Copy constructors
|
||||
BaseIPC(const BaseIPC &rhs) : remove_ipc_on_exit_(false) {}
|
||||
BaseIPC &operator=(const BaseIPC &rhs) {
|
||||
if (&rhs != this) {
|
||||
remove_ipc_on_exit_ = false;
|
||||
}
|
||||
return *this;
|
||||
}
|
||||
/// Move constructors
|
||||
BaseIPC(BaseIPC &&rhs) noexcept : remove_ipc_on_exit_(rhs.remove_ipc_on_exit_) { rhs.remove_ipc_on_exit_ = false; }
|
||||
BaseIPC &operator=(BaseIPC &&rhs) noexcept {
|
||||
if (&rhs != this) {
|
||||
remove_ipc_on_exit_ = rhs.remove_ipc_on_exit_;
|
||||
rhs.remove_ipc_on_exit_ = false;
|
||||
}
|
||||
return *this;
|
||||
}
|
||||
|
||||
protected:
|
||||
bool remove_ipc_on_exit_;
|
||||
};
|
||||
|
||||
/// \brief This wraps a shared message for the communication between processes. It is used primarily
|
||||
/// for starting and stopping a server.
|
||||
class SharedMessage : public BaseIPC {
|
||||
public:
|
||||
using queue_id_t = int;
|
||||
SharedMessage() : msg_qid_(-1) {}
|
||||
explicit SharedMessage(queue_id_t qid) : msg_qid_(qid) {}
|
||||
~SharedMessage() override;
|
||||
|
||||
/// Copy constructors
|
||||
SharedMessage(const SharedMessage &rhs) : BaseIPC(rhs), msg_qid_(rhs.msg_qid_) {}
|
||||
SharedMessage &operator=(const SharedMessage &rhs) {
|
||||
if (&rhs != this) {
|
||||
msg_qid_ = rhs.msg_qid_;
|
||||
BaseIPC::operator=(rhs);
|
||||
}
|
||||
return *this;
|
||||
}
|
||||
/// Move constructors
|
||||
SharedMessage(SharedMessage &&rhs) noexcept : BaseIPC(std::move(rhs)) {
|
||||
msg_qid_ = rhs.msg_qid_;
|
||||
rhs.msg_qid_ = -1;
|
||||
}
|
||||
SharedMessage &operator=(SharedMessage &&rhs) noexcept {
|
||||
if (&rhs != this) {
|
||||
msg_qid_ = rhs.msg_qid_;
|
||||
rhs.msg_qid_ = -1;
|
||||
BaseIPC::operator=(std::move(rhs));
|
||||
}
|
||||
return *this;
|
||||
}
|
||||
|
||||
/// Return the private id
|
||||
queue_id_t GetMsgQueueId() const { return msg_qid_; }
|
||||
|
||||
/// \brief Create a private message queue
|
||||
Status Create();
|
||||
|
||||
/// Send a Status object
|
||||
Status SendStatus(const Status &rc);
|
||||
|
||||
/// Retrieve a Status object
|
||||
Status ReceiveStatus(Status *rc);
|
||||
|
||||
private:
|
||||
queue_id_t msg_qid_;
|
||||
};
|
||||
|
||||
/// \brief This wraps a shared memory for the communication between processes. It is used primarily
|
||||
/// for transporting large tensor rows.
|
||||
class SharedMemory : public BaseIPC {
|
||||
public:
|
||||
using shm_key_t = int;
|
||||
using shm_id_t = int;
|
||||
SharedMemory() : shm_id_(-1), shm_key_(-1), shmat_addr_(nullptr) {}
|
||||
explicit SharedMemory(shm_key_t public_key) : shm_id_(-1), shm_key_(public_key), shmat_addr_(nullptr) {}
|
||||
~SharedMemory() override;
|
||||
/// Copy constructors
|
||||
SharedMemory(const SharedMemory &rhs)
|
||||
: BaseIPC(rhs), shm_id_(rhs.shm_id_), shm_key_(rhs.shm_key_), shmat_addr_(rhs.shmat_addr_) {}
|
||||
SharedMemory &operator=(const SharedMemory &rhs) {
|
||||
if (&rhs != this) {
|
||||
shm_id_ = rhs.shm_id_;
|
||||
shm_key_ = rhs.shm_key_;
|
||||
shmat_addr_ = rhs.shmat_addr_;
|
||||
BaseIPC::operator=(rhs);
|
||||
}
|
||||
return *this;
|
||||
}
|
||||
/// Move constructors
|
||||
SharedMemory(SharedMemory &&rhs) noexcept : BaseIPC(std::move(rhs)) {
|
||||
shm_id_ = rhs.shm_id_;
|
||||
shm_key_ = rhs.shm_key_;
|
||||
shmat_addr_ = rhs.shmat_addr_;
|
||||
rhs.shm_id_ = -1;
|
||||
rhs.shm_key_ = -1;
|
||||
rhs.shmat_addr_ = nullptr;
|
||||
}
|
||||
SharedMemory &operator=(SharedMemory &&rhs) noexcept {
|
||||
if (&rhs != this) {
|
||||
shm_id_ = rhs.shm_id_;
|
||||
shm_key_ = rhs.shm_key_;
|
||||
shmat_addr_ = rhs.shmat_addr_;
|
||||
rhs.shm_id_ = -1;
|
||||
rhs.shm_key_ = -1;
|
||||
rhs.shmat_addr_ = nullptr;
|
||||
BaseIPC::operator=(std::move(rhs));
|
||||
}
|
||||
return *this;
|
||||
}
|
||||
/// \brief Set the public key
|
||||
void SetPublicKey(key_t public_key) { shm_key_ = public_key; }
|
||||
|
||||
/// \brief This returns where we attach to the shared memory.
|
||||
/// \return Base address of the shared memory.
|
||||
const void *SharedMemoryBaseAddr() const { return shmat_addr_; }
|
||||
void *SharedMemoryBaseAddr() { return shmat_addr_; }
|
||||
|
||||
/// \brief Attach to shared memory
|
||||
/// \return Status object
|
||||
Status Attach();
|
||||
|
||||
/// Detach from shared memory
|
||||
/// \return Status object
|
||||
Status Detach();
|
||||
|
||||
/// Create shared memory
|
||||
/// \return Status object
|
||||
Status Create(int64_t sz);
|
||||
|
||||
/// Destroy shared memory
|
||||
/// \return Status object
|
||||
Status Destroy();
|
||||
|
||||
/// \brief Return the shared memory id
|
||||
shm_id_t GetSharedMemoryId() const { return shm_id_; }
|
||||
|
||||
/// \brief Get number of processes attached to the shared memory
|
||||
/// \return Status object
|
||||
Status GetNumAttached(int32_t *num);
|
||||
|
||||
private:
|
||||
shm_id_t shm_id_;
|
||||
shm_key_t shm_key_;
|
||||
void *shmat_addr_;
|
||||
};
|
||||
|
||||
/// \brief Generate a shared memory key using the tcp/ip port.
|
||||
/// \note It must be called after the cache server generates the unix socket or ftok will fail.
|
||||
/// \note Caller must check the return value. -1 means ftok failed.
|
||||
/// \param[in] port
|
||||
/// \param[out] err. If not null and ftok fails, this will contain the value of errno
|
||||
/// \return key
|
||||
Status PortToFtok(int port, SharedMemory::shm_key_t *);
|
||||
|
||||
} // namespace dataset
|
||||
} // namespace mindspore
|
||||
#endif // MINDSPORE_CCSRC_MINDDATA_DATASET_ENGINE_CACHE_IPC_H_
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
Some files were not shown because too many files have changed in this diff Show More
Loading…
Reference in new issue