frozenleaves/pytorch
1
0
Fork 0
mirror of https://github.com/pytorch/pytorch.git synced 2025-10-20 21:14:14 +08:00
Code Issues Packages Projects Releases Wiki Activity

[NOOP][clangformat][codemod] Enable CLANGFORMAT for some folders in caffe2/* (#67746)

Browse Source 
Summary: Pull Request resolved: https://github.com/pytorch/pytorch/pull/67746

Test Plan: Visual inspection. Sandcastle.

Reviewed By: zertosh

Differential Revision: D31986646

fbshipit-source-id: 91885c20c3cead3853c49abb9fe0a94a67f33cc8
This commit is contained in:
Shashank Chaudhry
2021-11-03 12:21:28 -07:00
committed by Facebook GitHub Bot
parent a5b57c9433
commit 89c4e8c22b
34 changed files with 937 additions and 712 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
torch/lib/libshm/core.cpp
View File

@ -4,8 +4,8 @@
#include <unordered_map>
#include <libshm/err.h>
#include <libshm/socket.h>
#include <libshm/libshm.h>
#include <libshm/socket.h>
std::unordered_map<std::string, ClientSocket> managers;
std::string manager_executable_path;
@ -47,7 +47,7 @@ void start_manager() {
constexpr auto MAX_BUFFER_SIZE = 1000;
std::array<char, MAX_BUFFER_SIZE> buffer;
std::string handle;
while(handle.empty() || handle.back() != '\n') {
while (handle.empty() || handle.back() != '\n') {
const auto bytes_read = read(pipe_ends[0], buffer.data(), buffer.size());
SYSCHECK_ERR_RETURN_NEG1(bytes_read);
if (bytes_read == 0) {
@ -68,11 +68,11 @@ void start_manager() {
std::string msg("torch_shm_manager at \"");
msg += manager_executable_path;
msg += "\": ";
msg += handle.substr(7); // remove "ERROR: "
msg += handle.substr(7); // remove "ERROR: "
throw std::runtime_error(msg);
}
ClientSocket manager {handle};
ClientSocket manager{handle};
managers.emplace(std::move(handle), std::move(manager));
}
@ -87,41 +87,49 @@ ClientSocket& get_manager_socket(const std::string& manager_handle) {
}
}
void libshm_init(const char *manager_exec_path) {
void libshm_init(const char* manager_exec_path) {
manager_executable_path = std::string(manager_exec_path);
}
THManagedMapAllocatorInit::THManagedMapAllocatorInit(const char* manager_handle, const char* filename)
: manager_handle_(manager_handle ? manager_handle : "") {
THManagedMapAllocatorInit::THManagedMapAllocatorInit(
const char* manager_handle,
const char* filename)
: manager_handle_(manager_handle ? manager_handle : "") {
// TODO: unlock GIL when contacting the manager
try {
// NOLINTNEXTLINE(cppcoreguidelines-init-variables)
ClientSocket *socket;
ClientSocket* socket;
if (!manager_handle_.empty()) {
socket = &get_manager_socket(manager_handle_);
} else {
if (managers.size() == 0) {
start_manager();
}
const auto &manager = managers.begin();
const auto& manager = managers.begin();
manager_handle_ = manager->first;
socket = &manager->second;
}
AllocInfo info = get_alloc_info(filename);
socket->register_allocation(info);
} catch(std::exception &e) {
} catch (std::exception& e) {
TORCH_CHECK(false, e.what());
}
}
THManagedMapAllocator::THManagedMapAllocator(const char *manager_handle, const char *filename, int flags, ptrdiff_t size)
: THManagedMapAllocatorInit(manager_handle, filename), at::RefcountedMapAllocator(filename, flags, size) {}
THManagedMapAllocator::THManagedMapAllocator(
const char* manager_handle,
const char* filename,
int flags,
ptrdiff_t size)
: THManagedMapAllocatorInit(manager_handle, filename),
at::RefcountedMapAllocator(filename, flags, size) {}
void THManagedMapAllocator::close() {
if (closed_) return;
if (closed_)
return;
AllocInfo info = get_alloc_info(filename());
info.free = true;
ClientSocket &socket = get_manager_socket(manager_handle_);
ClientSocket& socket = get_manager_socket(manager_handle_);
at::RefcountedMapAllocator::close();
socket.register_deallocation(info);
}
@ -130,11 +138,21 @@ static void deleteTHManagedMapAllocator(void* ptr) {
delete static_cast<THManagedMapAllocator*>(ptr);
}
at::DataPtr THManagedMapAllocator::makeDataPtr(const char* manager_handle, const char* filename, int flags, ptrdiff_t size) {
auto* context = new THManagedMapAllocator(manager_handle, filename, flags, size);
return {context->data(), context, &deleteTHManagedMapAllocator, at::DeviceType::CPU};
at::DataPtr THManagedMapAllocator::makeDataPtr(
const char* manager_handle,
const char* filename,
int flags,
ptrdiff_t size) {
auto* context =
new THManagedMapAllocator(manager_handle, filename, flags, size);
return {
context->data(),
context,
&deleteTHManagedMapAllocator,
at::DeviceType::CPU};
}
THManagedMapAllocator* THManagedMapAllocator::fromDataPtr(const at::DataPtr& dptr) {
THManagedMapAllocator* THManagedMapAllocator::fromDataPtr(
const at::DataPtr& dptr) {
return dptr.cast_context<THManagedMapAllocator>(&deleteTHManagedMapAllocator);
}

26
torch/lib/libshm/err.h
View File

@ -1,7 +1,7 @@
#pragma once
#include <system_error>
#include <cerrno>
#include <system_error>
// `errno` is only meaningful when it fails. E.g., a successful `fork()` sets
// `errno` to `EINVAL` in child process on some macos
@ -11,15 +11,15 @@
// All functions used in `libshm` (so far) indicate error by returning `-1`. If
// you want to use a function with a different error reporting mechanism, you
// need to port `SYSCHECK` from `torch/lib/c10d/Utils.hpp`.
#define SYSCHECK_ERR_RETURN_NEG1(expr) \
while (true) { \
if ((expr) == -1) { \
if (errno == EINTR) { \
continue; \
} else { \
throw std::system_error(errno, std::system_category()); \
} \
} else { \
break; \
} \
}
#define SYSCHECK_ERR_RETURN_NEG1(expr) \
while (true) { \
if ((expr) == -1) { \
if (errno == EINTR) { \
continue; \
} else { \
throw std::system_error(errno, std::system_category()); \
} \
} else { \
break; \
} \
}

29
torch/lib/libshm/libshm.h
View File

@ -4,11 +4,11 @@
#ifdef __cplusplus
void libshm_init(const char *manager_exec_path);
void libshm_init(const char* manager_exec_path);
// Superclass to run a constructor before at::RefcountedMapAllocator
class THManagedMapAllocatorInit {
protected:
protected:
THManagedMapAllocatorInit(const char* manager_handle, const char* filename);
std::string manager_handle_;
};
@ -16,18 +16,31 @@ protected:
// Like a at::RefcountedMapAllocator, but it also makes use of an external
// shared memory manager process to ensure that shared memory regions actually
// get freed in the end (even if processes lose the memory).
class THManagedMapAllocator : private THManagedMapAllocatorInit, public at::RefcountedMapAllocator {
public:
THManagedMapAllocator(const char* manager_handle, const char* filename, int flags, ptrdiff_t size);
class THManagedMapAllocator : private THManagedMapAllocatorInit,
public at::RefcountedMapAllocator {
public:
THManagedMapAllocator(
const char* manager_handle,
const char* filename,
int flags,
ptrdiff_t size);
void close() override;
~THManagedMapAllocator() { close(); }
~THManagedMapAllocator() {
close();
}
static at::DataPtr makeDataPtr(const char* manager_handle, const char* filename, int flags, ptrdiff_t size);
static at::DataPtr makeDataPtr(
const char* manager_handle,
const char* filename,
int flags,
ptrdiff_t size);
static THManagedMapAllocator* fromDataPtr(const at::DataPtr&);
const char* manager_handle() const { return manager_handle_.c_str(); }
const char* manager_handle() const {
return manager_handle_.c_str();
}
};
#endif

60
torch/lib/libshm/manager.cpp
View File

@ -1,11 +1,11 @@
#include <algorithm>
#include <cerrno>
#include <fcntl.h>
#include <memory>
#include <poll.h>
#include <set>
#include <sys/mman.h>
#include <unistd.h>
#include <algorithm>
#include <cerrno>
#include <memory>
#include <set>
#include <unordered_map>
#include <vector>
@ -26,19 +26,17 @@ const int SHUTDOWN_TIMEOUT = 2000; // 2s
#endif
struct ClientSession {
ClientSession(ManagerSocket s): socket(std::move(s)), pid(0) {}
ClientSession(ManagerSocket s) : socket(std::move(s)), pid(0) {}
ManagerSocket socket;
pid_t pid;
};
std::vector<struct pollfd> pollfds;
std::unordered_map<int, ClientSession> client_sessions;
// TODO: check if objects have been freed from time to time
std::set<std::string> used_objects;
void register_fd(int fd) {
struct pollfd pfd = {0};
pfd.fd = fd;
@ -46,17 +44,17 @@ void register_fd(int fd) {
pollfds.push_back(pfd);
}
void unregister_fd(int fd) {
pollfds.erase(
std::remove_if(pollfds.begin(), pollfds.end(),
[fd](const struct pollfd &pfd) { return pfd.fd == fd; }),
pollfds.end());
std::remove_if(
pollfds.begin(),
pollfds.end(),
[fd](const struct pollfd& pfd) { return pfd.fd == fd; }),
pollfds.end());
client_sessions.erase(fd);
}
void print_init_message(const char *message) {
void print_init_message(const char* message) {
// NOLINTNEXTLINE(cppcoreguidelines-init-variables)
size_t unused;
// NOLINTNEXTLINE(clang-analyzer-deadcode.DeadStores)
@ -65,7 +63,7 @@ void print_init_message(const char *message) {
unused = write(1, "\n", 1);
}
bool object_exists(const char *name) {
bool object_exists(const char* name) {
int fd = shm_open(name, O_RDONLY, 0);
if (fd >= 0) {
close(fd);
@ -75,7 +73,7 @@ bool object_exists(const char *name) {
}
}
void free_used_object(const std::string &name) {
void free_used_object(const std::string& name) {
if (!object_exists(name.c_str())) {
DEBUG("object %s appears to have been freed", name.c_str());
used_objects.erase(name);
@ -85,14 +83,13 @@ void free_used_object(const std::string &name) {
}
// NOLINTNEXTLINE(bugprone-exception-escape)
int main(int argc, char *argv[]) {
setsid(); // Daemonize the process
int main(int argc, char* argv[]) {
setsid(); // Daemonize the process
std::unique_ptr<ManagerServerSocket> srv_socket;
c10::optional<c10::TempDir> tempdir;
try {
tempdir =
c10::try_make_tempdir(/*name_prefix=*/"torch-shm-dir-");
tempdir = c10::try_make_tempdir(/*name_prefix=*/"torch-shm-dir-");
if (!tempdir.has_value()) {
throw std::runtime_error(
"could not generate a random directory for manager socket");
@ -122,17 +119,18 @@ int main(int argc, char *argv[]) {
int nevents;
if (client_sessions.size() == 0)
timeout = SHUTDOWN_TIMEOUT;
SYSCHECK_ERR_RETURN_NEG1(nevents = poll(pollfds.data(), pollfds.size(), timeout));
SYSCHECK_ERR_RETURN_NEG1(
nevents = poll(pollfds.data(), pollfds.size(), timeout));
timeout = -1;
if (nevents == 0 && client_sessions.size() == 0)
break;
for (auto &pfd: pollfds) {
for (auto& pfd : pollfds) {
if (pfd.revents & (POLLERR | POLLHUP)) {
// some process died
DEBUG("detaching process");
auto &session = client_sessions.at(pfd.fd);
(void) session;
auto& session = client_sessions.at(pfd.fd);
(void)session;
DEBUG("%d has died", session.pid);
to_remove.push_back(pfd.fd);
} else if (pfd.revents & POLLIN) {
@ -146,10 +144,14 @@ int main(int argc, char *argv[]) {
} else {
// someone wants to register a segment
DEBUG("got alloc info");
auto &session = client_sessions.at(pfd.fd);
auto& session = client_sessions.at(pfd.fd);
AllocInfo info = session.socket.receive();
session.pid = info.pid;
DEBUG("got alloc info: %d %d %s", (int)info.free, info.pid, info.filename);
DEBUG(
"got alloc info: %d %d %s",
(int)info.free,
info.pid,
info.filename);
if (info.free) {
free_used_object(info.filename);
} else {
@ -161,22 +163,22 @@ int main(int argc, char *argv[]) {
}
}
for (int fd: to_add)
for (int fd : to_add)
register_fd(fd);
to_add.clear();
for (int fd: to_remove)
for (int fd : to_remove)
unregister_fd(fd);
to_remove.clear();
}
for (auto &obj_name: used_objects) {
for (auto& obj_name : used_objects) {
DEBUG("freeing %s", obj_name.c_str());
shm_unlink(obj_name.c_str());
}
// Clean up file descriptors
for (auto &pfd: pollfds) {
for (auto& pfd : pollfds) {
unregister_fd(pfd.fd);
}
// Clean up manager.sock

90
torch/lib/libshm/socket.h
View File

@ -1,31 +1,33 @@
#pragma once
#include <sys/types.h>
#include <poll.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/un.h>
#include <unistd.h>
#include <poll.h>
#include <cstdio>
#include <string>
#include <sstream>
#include <iostream>
#include <cstring>
#include <cstddef>
#include <cstdio>
#include <cstring>
#include <iostream>
#include <sstream>
#include <string>
#include <libshm/err.h>
#include <libshm/alloc_info.h>
#include <libshm/err.h>
class Socket {
public:
public:
int socket_fd;
protected:
protected:
Socket() {
SYSCHECK_ERR_RETURN_NEG1(socket_fd = socket(AF_UNIX, SOCK_STREAM, 0));
}
Socket(const Socket& other) = delete;
Socket(Socket&& other) noexcept : socket_fd(other.socket_fd) { other.socket_fd = -1; };
Socket(Socket&& other) noexcept : socket_fd(other.socket_fd) {
other.socket_fd = -1;
};
explicit Socket(int fd) : socket_fd(fd) {}
virtual ~Socket() {
@ -33,7 +35,7 @@ protected:
close(socket_fd);
}
struct sockaddr_un prepare_address(const char *path) {
struct sockaddr_un prepare_address(const char* path) {
struct sockaddr_un address;
address.sun_family = AF_UNIX;
strcpy(address.sun_path, path);
@ -45,8 +47,8 @@ protected:
return offsetof(sockaddr_un, sun_path) + strlen(address.sun_path) + 1;
}
void recv(void *_buffer, size_t num_bytes) {
char *buffer = (char*)_buffer;
void recv(void* _buffer, size_t num_bytes) {
char* buffer = (char*)_buffer;
size_t bytes_received = 0;
ssize_t step_received;
struct pollfd pfd = {0};
@ -55,36 +57,39 @@ protected:
while (bytes_received < num_bytes) {
SYSCHECK_ERR_RETURN_NEG1(poll(&pfd, 1, 1000));
if (pfd.revents & POLLIN) {
SYSCHECK_ERR_RETURN_NEG1(step_received = ::read(socket_fd, buffer, num_bytes - bytes_received));
SYSCHECK_ERR_RETURN_NEG1(
step_received =
::read(socket_fd, buffer, num_bytes - bytes_received));
if (step_received == 0)
throw std::runtime_error("Other end has closed the connection");
bytes_received += step_received;
buffer += step_received;
} else if (pfd.revents & (POLLERR | POLLHUP)) {
throw std::runtime_error("An error occurred while waiting for the data");
throw std::runtime_error(
"An error occurred while waiting for the data");
} else {
throw std::runtime_error("Shared memory manager connection has timed out");
throw std::runtime_error(
"Shared memory manager connection has timed out");
}
}
}
void send(const void *_buffer, size_t num_bytes) {
const char *buffer = (const char*)_buffer;
void send(const void* _buffer, size_t num_bytes) {
const char* buffer = (const char*)_buffer;
size_t bytes_sent = 0;
ssize_t step_sent;
while (bytes_sent < num_bytes) {
SYSCHECK_ERR_RETURN_NEG1(step_sent = ::write(socket_fd, buffer, num_bytes));
SYSCHECK_ERR_RETURN_NEG1(
step_sent = ::write(socket_fd, buffer, num_bytes));
bytes_sent += step_sent;
buffer += step_sent;
}
}
};
class ManagerSocket: public Socket {
public:
explicit ManagerSocket(int fd): Socket(fd) {}
class ManagerSocket : public Socket {
public:
explicit ManagerSocket(int fd) : Socket(fd) {}
AllocInfo receive() {
AllocInfo info;
@ -95,20 +100,19 @@ public:
void confirm() {
send("OK", 2);
}
};
class ManagerServerSocket: public Socket {
public:
explicit ManagerServerSocket(const std::string &path) {
class ManagerServerSocket : public Socket {
public:
explicit ManagerServerSocket(const std::string& path) {
socket_path = path;
try {
struct sockaddr_un address = prepare_address(path.c_str());
size_t len = address_length(address);
SYSCHECK_ERR_RETURN_NEG1(bind(socket_fd, (struct sockaddr *)&address, len));
SYSCHECK_ERR_RETURN_NEG1(
bind(socket_fd, (struct sockaddr*)&address, len));
SYSCHECK_ERR_RETURN_NEG1(listen(socket_fd, 10));
} catch(std::exception &e) {
} catch (std::exception& e) {
SYSCHECK_ERR_RETURN_NEG1(close(socket_fd));
throw;
}
@ -128,36 +132,38 @@ public:
int client_fd;
struct sockaddr_un addr;
socklen_t addr_len = sizeof(addr);
SYSCHECK_ERR_RETURN_NEG1(client_fd = ::accept(socket_fd, (struct sockaddr *)&addr, &addr_len));
SYSCHECK_ERR_RETURN_NEG1(
client_fd = ::accept(socket_fd, (struct sockaddr*)&addr, &addr_len));
return ManagerSocket(client_fd);
}
std::string socket_path;
};
class ClientSocket: public Socket {
public:
explicit ClientSocket(const std::string &path) {
class ClientSocket : public Socket {
public:
explicit ClientSocket(const std::string& path) {
try {
struct sockaddr_un address = prepare_address(path.c_str());
size_t len = address_length(address);
SYSCHECK_ERR_RETURN_NEG1(connect(socket_fd, (struct sockaddr *)&address, len));
} catch(std::exception &e) {
SYSCHECK_ERR_RETURN_NEG1(
connect(socket_fd, (struct sockaddr*)&address, len));
} catch (std::exception& e) {
SYSCHECK_ERR_RETURN_NEG1(close(socket_fd));
throw;
}
}
void register_allocation(AllocInfo &info) {
void register_allocation(AllocInfo& info) {
char buffer[3] = {0, 0, 0};
send(&info, sizeof(info));
recv(buffer, 2);
if (strcmp(buffer, "OK") != 0)
throw std::runtime_error("Shared memory manager didn't respond with an OK");
throw std::runtime_error(
"Shared memory manager didn't respond with an OK");
}
void register_deallocation(AllocInfo &info) {
void register_deallocation(AllocInfo& info) {
send(&info, sizeof(info));
}
};

16
torch/lib/libshm_windows/core.cpp
View File

@ -4,19 +4,23 @@
#include <libshm_windows/libshm.h>
void libshm_init(const char *manager_exec_path) {
}
void libshm_init(const char* manager_exec_path) {}
static void deleteTHManagedMapAllocator(void* ptr) {
delete static_cast<THManagedMapAllocator*>(ptr);
}
at::DataPtr THManagedMapAllocator::makeDataPtr(const char* manager_handle, const char* filename, int flags, ptrdiff_t size) {
auto* context = new THManagedMapAllocator(manager_handle, filename, flags, size);
at::DataPtr THManagedMapAllocator::makeDataPtr(
const char* manager_handle,
const char* filename,
int flags,
ptrdiff_t size) {
auto* context =
new THManagedMapAllocator(manager_handle, filename, flags, size);
return {context->data(), context, &deleteTHManagedMapAllocator, at::kCPU};
}
THManagedMapAllocator* THManagedMapAllocator::fromDataPtr(const at::DataPtr& dptr) {
THManagedMapAllocator* THManagedMapAllocator::fromDataPtr(
const at::DataPtr& dptr) {
return dptr.cast_context<THManagedMapAllocator>(&deleteTHManagedMapAllocator);
}

26
torch/lib/libshm_windows/libshm.h
View File

@ -5,22 +5,32 @@
#ifdef __cplusplus
#ifdef SHM_EXPORTS
# define SHM_API __declspec(dllexport)
#define SHM_API __declspec(dllexport)
#else
# define SHM_API __declspec(dllimport)
#define SHM_API __declspec(dllimport)
#endif
SHM_API void libshm_init(const char *manager_exec_path);
SHM_API void libshm_init(const char* manager_exec_path);
class SHM_API THManagedMapAllocator : public at::RefcountedMapAllocator {
public:
THManagedMapAllocator(const char* manager_handle, const char* filename, int flags, ptrdiff_t size)
: at::RefcountedMapAllocator(filename, flags, size) {}
public:
THManagedMapAllocator(
const char* manager_handle,
const char* filename,
int flags,
ptrdiff_t size)
: at::RefcountedMapAllocator(filename, flags, size) {}
static at::DataPtr makeDataPtr(const char* manager_handle, const char* filename, int flags, ptrdiff_t size);
static at::DataPtr makeDataPtr(
const char* manager_handle,
const char* filename,
int flags,
ptrdiff_t size);
static THManagedMapAllocator* fromDataPtr(const at::DataPtr&);
const char* manager_handle() const { return "no_manager"; }
const char* manager_handle() const {
return "no_manager";
}
};
#endif