mmr/Livox-SDK
Archived
1
0
Fork 0
Code Issues 105 Pull Requests 11 Actions Packages Projects Releases 9 Wiki Activity

remove dependency on APR library

Browse Source
This commit is contained in:
Livox-SDK
2020-12-04 22:01:38 +08:00
parent eb19d42ce4
commit 7cf759a760
564 changed files with 2880 additions and 181270 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sdk_core/src/base/io_loop.cpp
+48 -106
View File
@@ -26,6 +26,7 @@
#include <functional>
#include <mutex>
#include <iostream>
#include <algorithm>
#include "logging.h"
using std::lock_guard;
@@ -33,101 +34,37 @@ using std::mutex;
using std::vector;
using std::chrono::steady_clock;
namespace livox {
bool IOLoop::Init() {
if (mem_pool_ == NULL) {
auto multiple_io = MultipleIOFactory::CreateMultipleIO();
if (!multiple_io) {
LOG_ERROR("Creat Multiple IO Failed!");
return false;
}
apr_status_t rv =
apr_pollset_create(&pollset_, kMaxPollCount, mem_pool_, APR_POLLSET_WAKEABLE);
if (rv != APR_SUCCESS) {
LOG_ERROR(PrintAPRStatus(rv));
multiple_io_base_ = std::move(multiple_io);
if (!multiple_io_base_->PollCreate(OPEN_MAX_POLL)) {
LOG_ERROR("Poll Create Failed!");
return false;
}
return true;
}
void IOLoop::Uninit() {
if (pollset_) {
apr_pollset_destroy(pollset_);
pollset_ = NULL;
}
for (DelegatesType::const_iterator ite = delegates_.begin(); ite != delegates_.end(); ++ite) {
ClientData *data = ite->second;
if (data) {
delete data;
}
}
delegates_.clear();
mem_pool_ = NULL;
multiple_io_base_->PollDestroy();
}
void IOLoop::AddDelegate(apr_socket_t *sock, IOLoop::IOLoopDelegate *delegate, void *data) {
void IOLoop::AddDelegate(socket_t sock, IOLoop::IOLoopDelegate *delegate, void *data) {
PostTask(std::bind(&IOLoop::AddDelegateAsync, this, sock, delegate, data));
}
void IOLoop::RemoveDelegate(apr_socket_t *sock, IOLoopDelegate *) {
void IOLoop::RemoveDelegate(socket_t sock, IOLoopDelegate *) {
PostTask(std::bind(&IOLoop::RemoveDelegateAsync, this, sock));
}
void IOLoop::RemoveDelegateSync(apr_socket_t *sock) {
assert(apr_os_thread_equal(this->thread_id_, apr_os_thread_current()));
RemoveDelegateAsync(sock);
}
void IOLoop::Loop() {
apr_int32_t num = 0;
const apr_pollfd_t *ret_pfd = NULL;
apr_status_t rv = apr_pollset_poll(pollset_, apr_time_from_msec(50), &num, &ret_pfd);
if (rv == APR_SUCCESS) {
for (int i = 0; i < num; i++) {
ClientData *data = static_cast<ClientData *>(ret_pfd[i].client_data);
if (data) {
IOLoopDelegate *delegate = data->first;
if (delegate) {
delegate->OnData(ret_pfd[i].desc.s, data->second);
}
}
}
}
if (enable_wake_ && APR_STATUS_IS_EINTR(rv)) {
DelegatesType delegates = delegates_;
for (DelegatesType::const_iterator ite = delegates.begin(); ite != delegates.end(); ++ite) {
ClientData *data = ite->second;
if (data) {
IOLoopDelegate *delegate = data->first;
if (delegate) {
delegate->OnWake();
}
}
}
}
if (enable_timer_) {
TimePoint t = steady_clock::now();;
if (last_timeout_ == TimePoint() || t - last_timeout_ > std::chrono::milliseconds(50)) {
last_timeout_ = t;
// copy delegates_ in case delegate is removed in callback.
DelegatesType delegates = delegates_;
for (DelegatesType::const_iterator ite = delegates.begin(); ite != delegates.end(); ++ite) {
ClientData *data = ite->second;
if (data) {
IOLoopDelegate *delegate = data->first;
if (delegate) {
delegate->OnTimer(t);
}
}
}
}
}
multiple_io_base_->Poll(POLL_TIMEOUT);
vector<IOLoopTask> tasks;
{
@@ -135,24 +72,17 @@ void IOLoop::Loop() {
tasks.swap(pending_tasks_);
}
for (vector<IOLoopTask>::iterator ite = tasks.begin(); ite != tasks.end(); ++ite) {
IOLoopTask &task = *ite;
for (auto &task : tasks) {
task();
}
if (rv != APR_SUCCESS && !APR_STATUS_IS_TIMEUP(rv) && !APR_STATUS_IS_EINTR(rv)) {
LOG_ERROR(PrintAPRStatus(rv));
}
}
bool IOLoop::Wakeup() {
apr_status_t rv = apr_pollset_wakeup(pollset_);
if (rv != APR_SUCCESS) {
LOG_ERROR(PrintAPRStatus(rv));
return false;
bool IOLoop::Wakeup() {
if (multiple_io_base_) {
multiple_io_base_->PollWakeUp();
}
return true;
}
}
void IOLoop::PostTask(const IOLoopTask &task) {
{
@@ -162,27 +92,39 @@ void IOLoop::PostTask(const IOLoopTask &task) {
Wakeup();
}
void IOLoop::AddDelegateAsync(apr_socket_t *sock, IOLoop::IOLoopDelegate *delegate, void *data) {
ClientData *p = new ClientData(delegate, data);
apr_pollfd_t pfd = {mem_pool_, APR_POLL_SOCKET, APR_POLLIN, 0, {NULL}, p};
pfd.desc.s = sock;
apr_pollset_add(pollset_, &pfd);
delegates_[sock] = p;
void IOLoop::AddDelegateAsync(socket_t sock, IOLoop::IOLoopDelegate *delegate, void *data) {
PollFd pollfd = {};
pollfd.fd = sock;
pollfd.event = READBLE_EVENT;
pollfd.event_callback = [=](FdEvent event) {
if (event & READBLE_EVENT) {
if (delegate) {
delegate->OnData(sock, data);
}
}
};
if (enable_timer_) {
pollfd.timer_callback = [=](TimePoint t) {
if (delegate) {
delegate->OnTimer(t);
}
};
}
if (enable_wake_) {
pollfd.wake_callback = [=]() {
if (delegate) {
delegate->OnWake();
}
};
}
multiple_io_base_->PollSetAdd(pollfd);
}
void IOLoop::RemoveDelegateAsync(apr_socket_t *sock) {
apr_pollfd_t pfd = {mem_pool_, APR_POLL_SOCKET, 0, 0, {NULL}, NULL};
pfd.desc.s = sock;
apr_pollset_remove(pollset_, &pfd);
DelegatesType::iterator ite = delegates_.find(sock);
if (ite != delegates_.end()) {
ClientData *p = ite->second;
if (p) {
delete p;
}
delegates_.erase(ite);
}
void IOLoop::RemoveDelegateAsync(socket_t sock) {
PollFd pollfd = {};
pollfd.fd = sock;
pollfd.event = READBLE_EVENT;
multiple_io_base_->PollSetRemove(pollfd);
}
} // namespace livox
sdk_core/src/base/io_loop.h
+19 -27
View File
@@ -30,61 +30,53 @@
#include <unordered_map>
#include <utility>
#include <vector>
#include "apr_general.h"
#include "apr_network_io.h"
#include "apr_poll.h"
#include "apr_thread_proc.h"
#include <algorithm>
#include "command_callback.h"
#include "noncopyable.h"
#include "thread_base.h"
#include "util.h"
#include "multiple_io/multiple_io_base.h"
#include "multiple_io/multiple_io_factory.h"
namespace livox {
#define OPEN_MAX_POLL 48
#define POLL_TIMEOUT 50 //ms
typedef int socket_t;
class IOLoop : public noncopyable {
public:
typedef std::chrono::steady_clock::time_point TimePoint;
typedef std::function<void(void)> IOLoopTask;
class IOLoopDelegate {
public:
virtual void OnData(apr_socket_t *, void *) {}
virtual void OnTimer(TimePoint) {}
virtual void OnData(socket_t, void *) {}
virtual void OnTimer(std::chrono::steady_clock::time_point) {}
virtual void OnWake() {}
};
typedef std::function<void(void)> IOLoopTask;
public:
explicit IOLoop(apr_pool_t *mem_pool, bool enable_timer = true, bool enable_wake = true)
: pollset_(NULL), mem_pool_(mem_pool), last_timeout_(), enable_timer_(enable_timer), enable_wake_(enable_wake){};
explicit IOLoop(bool enable_timer = true, bool enable_wake = true)
: enable_timer_(enable_timer), enable_wake_(enable_wake){};
bool Init();
void Uninit();
void AddDelegate(apr_socket_t *sock, IOLoopDelegate *delegate, void *data = NULL);
void RemoveDelegate(apr_socket_t *sock, IOLoopDelegate *delegate);
void RemoveDelegateSync(apr_socket_t *sock);
void AddDelegate(socket_t sock, IOLoopDelegate *delegate, void *data = NULL);
void RemoveDelegate(socket_t sock, IOLoopDelegate *delegate);
void Loop();
bool Wakeup();
void PostTask(const IOLoopTask &task);
void SetThreadId (apr_os_thread_t thread_id) { thread_id_ = thread_id; }
apr_os_thread_t GetThreadId () { return thread_id_; }
private:
void AddDelegateAsync(apr_socket_t *sock, IOLoopDelegate *delegate, void *data);
void RemoveDelegateAsync(apr_socket_t *sock);
void AddDelegateAsync(socket_t sock, IOLoopDelegate *delegate, void *data);
void RemoveDelegateAsync(socket_t sock);
private:
static const apr_uint32_t kMaxPollCount = 48;
typedef std::pair<IOLoopDelegate *, void *> ClientData;
typedef std::unordered_map<apr_socket_t *, ClientData *> DelegatesType;
DelegatesType delegates_;
apr_pollset_t *pollset_;
apr_pool_t *mem_pool_;
TimePoint last_timeout_;
std::mutex mutex_;
bool enable_timer_;
bool enable_wake_;
std::vector<IOLoopTask> pending_tasks_;
apr_os_thread_t thread_id_;
std::unique_ptr<MultipleIOBase> multiple_io_base_;
};
} // namespace livox
sdk_core/src/base/io_thread.cpp
+2 -6
View File
@@ -23,7 +23,6 @@
//
#include "io_thread.h"
#include "apr_pools.h"
namespace livox {
@@ -32,11 +31,9 @@ void IOThread::Join() {
}
void IOThread::ThreadFunc() {
if (loop_ == NULL) {
if (!loop_) {
return;
}
apr_os_thread_t thread_id = apr_os_thread_current();
loop_->SetThreadId(thread_id);
while (!IsQuit()) {
loop_->Loop();
@@ -48,14 +45,13 @@ bool IOThread::Init(bool enable_timer, bool enable_wake) {
return false;
}
loop_.reset(new IOLoop(pool_, enable_timer, enable_wake));
loop_ = std::make_shared<IOLoop>(enable_timer, enable_wake);
return loop_->Init();
}
void IOThread::Uninit() {
if (loop_) {
loop_->Uninit();
loop_.reset(NULL);
}
ThreadBase::Uninit();
}
sdk_core/src/base/io_thread.h
+2 -2
View File
@@ -37,11 +37,11 @@ class IOThread : public ThreadBase {
void Join();
void Uninit();
IOLoop *loop() { return loop_.get(); }
std::weak_ptr<IOLoop> loop() { return loop_; }
void ThreadFunc();
private:
std::unique_ptr<IOLoop> loop_;
std::shared_ptr<IOLoop> loop_;
};
} // namespace livox
#endif // LIVOX_IO_THREAD_H_
sdk_core/src/base/multiple_io/multiple_io_base.cpp
+84
View File
@@ -0,0 +1,84 @@
//
// The MIT License (MIT)
//
// Copyright (c) 2019 Livox. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
#include "multiple_io_base.h"
namespace livox {
void MultipleIOBase::CheckTimer() {
TimePoint t = std::chrono::steady_clock::now();
if (t - last_timeout_ > std::chrono::milliseconds(50)) {
last_timeout_ = t;
for (auto & descriptor : descriptors_) {
PollFd pollfd = descriptor.second;
if (pollfd.timer_callback) {
pollfd.timer_callback(t);
}
}
}
}
void MultipleIOBase::WakeUpInit() {
//Initialize wake up pipe
wake_up_pipe_.reset(new WakeUpPipe());
wake_up_pipe_->PipeCreate();
//register wake_fd to multiple io
PollFd wake_fd = {};
wake_fd.fd = wake_up_pipe_->GetPipeOut();
wake_fd.event = READBLE_EVENT;
wake_fd.event_callback = [this](FdEvent event) {
if (event & READBLE_EVENT) {
if (wake_up_pipe_) {
wake_up_pipe_->Drain();
}
for (auto & descriptor : descriptors_) {
PollFd pollfd = descriptor.second;
if (pollfd.wake_callback) {
pollfd.wake_callback();
}
}
}
};
PollSetAdd(wake_fd);
}
void MultipleIOBase::WakeUpUninit() {
PollFd wake_fd = {};
wake_fd.fd = wake_up_pipe_->GetPipeOut();
wake_fd.event = READBLE_EVENT | WRITABLE_EVENT;
PollSetRemove(wake_fd);
if (wake_up_pipe_) {
wake_up_pipe_->PipeDestroy();
wake_up_pipe_ = nullptr;
}
}
void MultipleIOBase::PollWakeUp() {
if (wake_up_pipe_) {
wake_up_pipe_->WakeUp();
}
return;
}
} // namespace livox
sdk_core/src/base/multiple_io/multiple_io_base.h
+73
View File
@@ -0,0 +1,73 @@
//
// The MIT License (MIT)
//
// Copyright (c) 2019 Livox. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
#ifndef MULTIPLE_IO_BASE_H_
#define MULTIPLE_IO_BASE_H_
#include <map>
#include <functional>
#include <chrono>
#include <memory>
#include "base/wake_up/wake_up_pipe.h"
namespace livox {
#define NONE_EVENT 0 /* No events registered. */
#define READBLE_EVENT 1 /* when descriptor is readable. */
#define WRITABLE_EVENT 2 /* when descriptor is writeable. */
typedef std::chrono::steady_clock::time_point TimePoint;
typedef int FdEvent;
typedef struct {
int fd; /* File descriptor. */
FdEvent event; /* Read | Write Event to listen. */
std::function<void(FdEvent)> event_callback; /* Read or Write Event Callback. */
std::function<void(TimePoint)> timer_callback; /* Timer Event Callback. */
std::function<void()> wake_callback; /* WakeUp Event Callback. */
} PollFd;
class MultipleIOBase {
public:
MultipleIOBase() = default;
virtual ~MultipleIOBase() = default;
virtual bool PollCreate(int size) = 0;
virtual void PollDestroy() = 0;
virtual bool PollSetAdd(PollFd poll_fd) = 0;
virtual bool PollSetRemove(PollFd poll_fd) = 0;
virtual void Poll(int timeout) = 0;
virtual void PollWakeUp();
protected:
virtual void CheckTimer();
std::map<int, PollFd> descriptors_;
TimePoint last_timeout_ = TimePoint();
virtual void WakeUpInit();
virtual void WakeUpUninit();
std::unique_ptr<WakeUpPipe> wake_up_pipe_;
};
} // namespace livox
#endif // MULTIPLE_IO_BASE_H_
sdk_core/src/base/multiple_io/multiple_io_epoll.cpp
+107
View File
@@ -0,0 +1,107 @@
//
// The MIT License (MIT)
//
// Copyright (c) 2019 Livox. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
#include "multiple_io_epoll.h"
#ifdef HAVE_EPOLL
namespace livox {
int GetEvent (FdEvent event) {
FdEvent rv = 0;
if (event & READBLE_EVENT)
rv |= EPOLLIN;
if (event & WRITABLE_EVENT)
rv |= EPOLLOUT;
return rv;
}
bool MultipleIOEpoll::PollCreate(int size) {
max_poll_size_ = size + 1;
epoll_fd_ = epoll_create(max_poll_size_);
if (epoll_fd_ < 0) {
return false;
}
pollset_.reset(new struct epoll_event[size]);
WakeUpInit();
return true;
}
void MultipleIOEpoll::PollDestroy() {
WakeUpUninit();
if (epoll_fd_ > 0) {
close(epoll_fd_);
epoll_fd_ = -1;
}
}
bool MultipleIOEpoll::PollSetAdd(PollFd poll_fd) {
if (max_poll_size_ <= (int)descriptors_.size()) {
return false;
}
struct epoll_event ee = {0};
ee.events = GetEvent(poll_fd.event);
ee.data.fd = poll_fd.fd;
if (epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, poll_fd.fd, &ee) == -1) {
return false;
}
int fd = poll_fd.fd;
descriptors_[fd] = poll_fd;
return true;
}
bool MultipleIOEpoll::PollSetRemove(PollFd poll_fd) {
int fd = poll_fd.fd;
struct epoll_event ee = {0};
epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, &ee);
if (descriptors_.find(fd) != descriptors_.end()) {
descriptors_.erase(fd);
}
return true;
}
void MultipleIOEpoll::Poll(int time_out) {
int ret = epoll_wait(epoll_fd_, pollset_.get(), (int)descriptors_.size(),
time_out);
if (ret > 0) {
for (int i =0; i< ret; i++) {
FdEvent fd_event = NONE_EVENT;
if (pollset_[i].events & EPOLLIN) {
fd_event |= READBLE_EVENT;
}
if (pollset_[i].events & EPOLLOUT) {
fd_event |= WRITABLE_EVENT;
}
int fd = pollset_[i].data.fd;
if (descriptors_.find(fd) != descriptors_.end()) {
PollFd pollfd = descriptors_[fd];
pollfd.event_callback(fd_event);
}
}
}
CheckTimer();
}
} // namespace livox
#endif // HAVE_EPOLL
sdk_core/src/base/multiple_io/multiple_io_epoll.h
+51
View File
@@ -0,0 +1,51 @@
//
// The MIT License (MIT)
//
// Copyright (c) 2019 Livox. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
#ifndef MULTIPLE_IO_EPOLL_H_
#define MULTIPLE_IO_EPOLL_H_
#include "multiple_io_base.h"
#include "config.h"
#include <memory>
#ifdef HAVE_EPOLL
namespace livox {
class MultipleIOEpoll : public MultipleIOBase {
public:
bool PollCreate(int size);
bool PollSetAdd(PollFd poll_fd);
bool PollSetRemove(PollFd poll_fd);
void Poll(int timeout);
void PollDestroy();
private:
int epoll_fd_ = -1;
std::unique_ptr<struct epoll_event[]> pollset_;
int max_poll_size_ = 0;
};
} // namespace livox
#endif // HAVE_EPOLL
#endif // MULTIPLE_IO_EPOLL_H_
sdk_core/src/base/multiple_io/multiple_io_factory.h
+56
View File
@@ -0,0 +1,56 @@
//
// The MIT License (MIT)
//
// Copyright (c) 2019 Livox. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
#ifndef MULTIPLE_IO_FACTORY_H_
#define MULTIPLE_IO_FACTORY_H_
#include "multiple_io_base.h"
#include "multiple_io_epoll.h"
#include "multiple_io_kqueue.h"
#include "multiple_io_select.h"
#include "multiple_io_poll.h"
#include <memory>
namespace livox {
class MultipleIOFactory {
public:
static std::unique_ptr<MultipleIOBase> CreateMultipleIO() {
#if defined(HAVE_EPOLL)
return std::unique_ptr<MultipleIOBase>(new MultipleIOEpoll());
#elif defined(HAVE_KQUEUE)
return std::unique_ptr<MultipleIOBase>(new MultipleIOKqueue());
#elif defined(HAVE_SELECT)
return std::unique_ptr<MultipleIOBase>(new MultipleIOSelect());
#elif defined(HAVE_POLL)
return std::unique_ptr<MultipleIOBase>(new MultipleIOPoll());
#else
return nullptr;
#endif
}
};
} // namespace livox
#endif // MULTIPLE_IO_FACTORY_H_
sdk_core/src/base/multiple_io/multiple_io_kqueue.cpp
+135
View File
@@ -0,0 +1,135 @@
//
// The MIT License (MIT)
//
// Copyright (c) 2019 Livox. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
#include "multiple_io_kqueue.h"
#ifdef HAVE_KQUEUE
namespace livox {
bool MultipleIOKqueue::PollCreate(int size) {
kqueue_fd_ = kqueue();
if (kqueue_fd_ == -1) {
return false;
}
int flags = 0;
if ((flags = fcntl(kqueue_fd_, F_GETFD)) == -1) {
close(kqueue_fd_);
return false;
}
flags |= FD_CLOEXEC;
if (fcntl(kqueue_fd_, F_SETFD, flags) == -1) {
close(kqueue_fd_);
return false;
}
max_poll_size_ = size + 1;
set_size_ = 2 * max_poll_size_;
kevent_set_.reset(new struct kevent[set_size_]);
WakeUpInit();
return true;
}
void MultipleIOKqueue::PollDestroy() {
WakeUpUninit();
if (kqueue_fd_ > 0) {
close(kqueue_fd_);
kqueue_fd_ = -1;
}
}
bool MultipleIOKqueue::PollSetAdd(PollFd poll_fd) {
if (max_poll_size_ <= (int)descriptors_.size()) {
return false;
}
int fd = poll_fd.fd;
descriptors_[fd] = poll_fd;
if (poll_fd.event & READBLE_EVENT) {
EV_SET(&kevent_, fd, EVFILT_READ, EV_ADD, 0, 0, &descriptors_[fd].fd);
if (kevent(kqueue_fd_, &kevent_, 1, nullptr, 0, nullptr) == -1) {
descriptors_.erase(fd);
return false;
}
}
if (poll_fd.event & WRITABLE_EVENT) {
EV_SET(&kevent_, fd, EVFILT_WRITE, EV_ADD, 0, 0, &descriptors_[fd].fd);
if (kevent(kqueue_fd_, &kevent_, 1, nullptr, 0, nullptr) == -1) {
descriptors_.erase(fd);
return false;
}
}
return true;
}
bool MultipleIOKqueue::PollSetRemove(PollFd poll_fd) {
int fd = poll_fd.fd;
if (descriptors_.find(fd) != descriptors_.end()) {
if (descriptors_[fd].event & READBLE_EVENT) {
EV_SET(&kevent_, fd, EVFILT_READ, EV_DELETE, 0, 0, NULL);
if (kevent(kqueue_fd_, &kevent_, 1, nullptr, 0, nullptr) == -1) {
return false;
}
}
if (descriptors_[fd].event & WRITABLE_EVENT) {
EV_SET(&kevent_, fd, EVFILT_WRITE, EV_DELETE, 0, 0, NULL);
if (kevent(kqueue_fd_, &kevent_, 1, nullptr, 0, nullptr) == -1) {
return false;
}
}
descriptors_.erase(fd);
}
return true;
}
void MultipleIOKqueue::Poll(int time_out) {
struct timespec tv, *tvptr;
if (time_out < 0) {
tvptr = NULL;
} else {
tv.tv_sec = (long) time_out / 1000;
tv.tv_nsec = (long) (time_out % 1000) * 1000000;
tvptr = &tv;
}
int rv = kevent(kqueue_fd_, NULL, 0, kevent_set_.get(), set_size_, tvptr);
if (rv > 0) {
for (int i = 0; i < rv; i++) {
int fd = *(int *)(kevent_set_[i].udata);
if (descriptors_.find(fd) != descriptors_.end()) {
PollFd pollfd = descriptors_[fd];
if (kevent_set_[i].filter == EVFILT_READ) {
pollfd.event_callback(READBLE_EVENT);
}
if (kevent_set_[i].filter == EVFILT_WRITE) {
pollfd.event_callback(WRITABLE_EVENT);
}
}
}
}
CheckTimer();
}
} // namespace livox
#endif // HAVE_KQUEUE
sdk_core/src/base/multiple_io/multiple_io_kqueue.h
+52
View File
@@ -0,0 +1,52 @@
//
// The MIT License (MIT)
//
// Copyright (c) 2019 Livox. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
#ifndef MULTIPLE_IO_KQUEUE_H_
#define MULTIPLE_IO_KQUEUE_H_
#include "multiple_io_base.h"
#include "config.h"
#ifdef HAVE_KQUEUE
namespace livox {
class MultipleIOKqueue : public MultipleIOBase {
public:
bool PollCreate(int size);
bool PollSetAdd(PollFd poll_fd);
bool PollSetRemove(PollFd poll_fd);
void Poll(int timeout);
void PollDestroy();
private:
int kqueue_fd_ = -1;
struct kevent kevent_ = {};
std::unique_ptr<struct kevent[]> kevent_set_;
int max_poll_size_ = 0;
int set_size_ = 0;
};
} // namespace livox
#endif // HAVE_KQUEUE
#endif // MULTIPLE_IO_KQUEUE_H_
sdk_core/src/base/multiple_io/multiple_io_poll.cpp
+114
View File
@@ -0,0 +1,114 @@
//
// The MIT License (MIT)
//
// Copyright (c) 2019 Livox. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
#include "multiple_io_poll.h"
#ifdef HAVE_POLL
namespace livox {
int GetEvent (FdEvent event) {
int rv = 0;
if (event & READBLE_EVENT)
rv |= POLLIN;
if (event & WRITABLE_EVENT)
rv |= POLLOUT;
return rv;
}
bool MultipleIOPoll:: PollCreate(int size) {
max_poll_size_ = size + 1;
pollset_.reset(new struct pollfd[max_poll_size_]);
WakeUpInit();
return true;
}
bool MultipleIOPoll:: PollSetAdd(PollFd poll_fd) {
if (max_poll_size_ <= (int)descriptors_.size()) {
return false;
}
int fd = poll_fd.fd;
descriptors_[fd] = poll_fd;
struct pollfd fds;
fds.fd = fd;
fds.events = GetEvent(poll_fd.event);
pollset_[pollset_num_] = fds;
pollset_num_++;
return true;
}
void MultipleIOPoll::PollDestroy() {
WakeUpUninit();
return;
}
bool MultipleIOPoll:: PollSetRemove(PollFd poll_fd) {
int fd = poll_fd.fd;
if (descriptors_.find(fd) != descriptors_.end()) {
descriptors_.erase(fd);
}
for (int i = 0; i< pollset_num_; i++) {
if (pollset_[i].fd == fd) {
int dst = i;
for (i++; i < pollset_num_ - 1; i++) {
if (pollset_[i].fd == fd) {
pollset_num_--;
} else {
pollset_[dst] = pollset_[i];
dst++;
}
}
}
}
return true;
}
void MultipleIOPoll:: Poll(int time_out) {
int rv = poll(pollset_.get(), pollset_num_, time_out);
if (rv > 0) {
for (int i = 0; i < pollset_num_; i++) {
FdEvent fd_event = NONE_EVENT;
if (pollset_[i].revents & POLLIN) {
fd_event = | READBLE_EVENT;
}
if (pollset_[i].revents & POLLOUT) {
fd_event = | WRITABLE_EVENT;
}
int fd = pollset_[i].fd;
if (descriptors_.find(fd) != descriptors_.end()) {
PollFd pollfd = descriptors_[fd];
pollfd.event_callback(fd_event);
}
pollset_[i].revents = NONE_EVENT;
}
}
CheckTimer();
}
} // namespace livox
#endif // HAVE_POLL
sdk_core/src/base/util.cpp → sdk_core/src/base/multiple_io/multiple_io_poll.h
+25 -24
View File
@@ -22,31 +22,32 @@
// SOFTWARE.
//
#include "util.h"
#include <stdio.h>
#include <stdlib.h>
using std::string;
#ifndef MULTIPLE_IO_POLL_H_
#define MULTIPLE_IO_POLL_H_
#include "multiple_io_base.h"
#include "config.h"
#include <memory>
#ifdef HAVE_POLL
namespace livox {
string PrintAPRStatus(apr_status_t s) {
char buf[128];
apr_strerror(s, buf, sizeof(buf));
return buf;
}
string PrintAPRTime(apr_time_t t) {
char cbuf[45 + 1];
apr_size_t sz = 45;
apr_time_exp_t xt;
apr_time_exp_gmt(&xt, t);
int milli = t % 1000;
apr_strftime(cbuf, &sz, 45, " %T.", &xt);
string s(cbuf);
char buffer[33];
sprintf(buffer, "%d", milli);
s += buffer;
return s;
}
class MultipleIOPoll : public MultipleIOBase {
public:
bool PollCreate(int size);
bool PollSetAdd(PollFd poll_fd);
bool PollSetRemove(PollFd poll_fd);
void Poll(int timeout);
void PollDestroy();
private:
std::unique_ptr<struct pollfd[]> pollset_;
int pollset_num_ = 0;
int max_poll_size_ = 0;
};
} // namespace livox
#endif // HAVE_POLL
#endif // MULTIPLE_IO_POLL_H_
sdk_core/src/base/multiple_io/multiple_io_select.cpp
+121
View File
@@ -0,0 +1,121 @@
//
// The MIT License (MIT)
//
// Copyright (c) 2019 Livox. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
#include "multiple_io_select.h"
#include <thread>
#ifdef HAVE_SELECT
namespace livox {
bool MultipleIOSelect:: PollCreate(int size) {
FD_ZERO(&rfds_);
FD_ZERO(&wfds_);
//wake up fd + 1
max_poll_size_ = size + 1;
WakeUpInit();
return true;
}
void MultipleIOSelect::PollDestroy() {
WakeUpUninit();
FD_ZERO(&rfds_);
FD_ZERO(&wfds_);
}
bool MultipleIOSelect:: PollSetAdd(PollFd poll_fd) {
if (max_poll_size_ <= (int)descriptors_.size()) {
return false;
}
int fd = poll_fd.fd;
descriptors_[fd] = poll_fd;
if (max_fd_ < fd) {
max_fd_ = fd;
}
if (poll_fd.event & READBLE_EVENT) {
FD_SET(fd, &rfds_);
}
if (poll_fd.event & WRITABLE_EVENT) {
FD_SET(fd, &wfds_);
}
return true;
}
bool MultipleIOSelect:: PollSetRemove(PollFd poll_fd) {
int fd = poll_fd.fd;
if (descriptors_.find(fd) != descriptors_.end()) {
descriptors_.erase(fd);
}
FD_CLR(fd, &rfds_);
FD_CLR(fd, &wfds_);
if (max_fd_ <= fd) {
max_fd_--;
}
return true;
}
void MultipleIOSelect:: Poll(int time_out) {
fd_set readset, writeset;
struct timeval tv, *tvptr;
if (descriptors_.size() == 0) {
if (time_out > 0) {
std::this_thread::sleep_for(std::chrono::milliseconds(time_out));
return;
}
return ;
}
if (time_out < 0) {
tvptr = nullptr;
} else {
tv.tv_sec = (long)time_out / 1000;
tv.tv_usec = (long)time_out % 1000;
tvptr = &tv;
}
memcpy(&readset, &rfds_, sizeof(fd_set));
memcpy(&writeset, &wfds_, sizeof(fd_set));
int rv = select(max_fd_ + 1, &readset, &writeset, nullptr, tvptr);
if (rv > 0) {
for (auto& descriptor : descriptors_) {
int fd = descriptor.first;
FdEvent fd_event = NONE_EVENT;
if (FD_ISSET(fd, &readset)) {
fd_event |= READBLE_EVENT;
}
if (FD_ISSET(fd, &writeset)) {
fd_event |= WRITABLE_EVENT;
}
if (fd_event != NONE_EVENT) {
PollFd pollfd = descriptor.second;
pollfd.event_callback(fd_event);
}
}
}
CheckTimer();
}
} // namespace livox
#endif // HAVE_SELECT
sdk_core/src/base/multiple_io/multiple_io_select.h
+52
View File
@@ -0,0 +1,52 @@
//
// The MIT License (MIT)
//
// Copyright (c) 2019 Livox. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
#ifndef MULTIPLE_IO_SELECT_H_
#define MULTIPLE_IO_SELECT_H_
#include "multiple_io_base.h"
#include "config.h"
#ifdef HAVE_SELECT
namespace livox {
class MultipleIOSelect : public MultipleIOBase {
public:
bool PollCreate(int size);
bool PollSetAdd(PollFd poll_fd);
bool PollSetRemove(PollFd poll_fd);
void Poll(int timeout);
void PollDestroy();
private:
int max_fd_ = -1;
fd_set rfds_;
fd_set wfds_;
int max_poll_size_ = 0;
};
} // namespace livox
#endif // HAVE_SELECT
#endif // MULTIPLE_IO_SELECT_H_
sdk_core/src/base/network_util.h → sdk_core/src/base/network/network_util.h
+21 -5
View File
@@ -24,19 +24,35 @@
#ifndef LIVOX_NETWORK_UTIL_H_
#define LIVOX_NETWORK_UTIL_H_
#include <apr_general.h>
#include <apr_network_io.h>
#ifdef WIN32
#include <stdint.h>
#endif
#ifdef WIN32
#include <winsock2.h>
#else
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#endif // WIN32
#include <stdio.h>
#include <stdio.h>
#include <fcntl.h>
namespace livox {
namespace util {
apr_socket_t *CreateBindSocket(uint16_t port, apr_pool_t *mem_pool, bool reuse_port = false, bool nonblock = true);
typedef int socket_t;
socket_t CreateSocket(uint16_t port, bool nonblock = true, bool reuse_port = true);
void CloseSock(socket_t sock);
bool FindLocalIp(const struct sockaddr_in &client_addr, uint32_t &local_ip);
size_t RecvFrom(socket_t &sock, void *buff, size_t buf_size, int flag, struct sockaddr *addr, int* addrlen);
} // namespace util
} // namespace livox
#endif // LIVOX_NETWORK_UTIL_H_
sdk_core/src/base/network/unix/network_util.cpp
+124
View File
@@ -0,0 +1,124 @@
//
// The MIT License (MIT)
//
// Copyright (c) 2019 Livox. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
#ifndef WIN32
#include "base/network/network_util.h"
#include <ifaddrs.h>
#include <string>
#include <string.h>
#include <sys/ioctl.h>
#include <unistd.h>
#include <netdb.h>
namespace livox {
namespace util {
socket_t CreateSocket(uint16_t port, bool nonblock, bool reuse_port) {
int status = -1;
int on = -1;
int sock = -1;
struct sockaddr_in servaddr;
sock = socket(AF_INET, SOCK_DGRAM, 0);
if (sock < 0) {
return -1;
}
if (nonblock) {
status = ioctl(sock, FIONBIO, (char*)&on);
if (status != 0) {
close(sock);
return -1;
}
}
if (reuse_port) {
status = setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
(char *) &on, sizeof (on));
if (status != 0) {
close(sock);
return -1;
}
}
memset(&servaddr, 0, sizeof(servaddr));
// Filling server information
servaddr.sin_family = AF_INET; // IPv4
servaddr.sin_addr.s_addr = INADDR_ANY;
servaddr.sin_port = htons(port);
status = bind(sock, (const struct sockaddr *)&servaddr, sizeof(servaddr));
if (status != 0) {
close(sock);
return -1;
}
return sock;
}
void CloseSock(int sock) {
if (sock > 0) {
close(sock);
}
}
bool FindLocalIp(const struct sockaddr_in &client_addr, uint32_t &local_ip) {
struct ifaddrs *if_addrs = NULL, *addrs = NULL;
if (getifaddrs(&if_addrs) == -1) {
return false;
}
addrs = if_addrs;
bool found = false;
while (if_addrs != NULL) {
if ((if_addrs->ifa_addr != NULL) && (if_addrs->ifa_addr->sa_family == AF_INET)) // check it is IP4
{
// is a connected IP4 Address
struct sockaddr_in *ifu_localaddr = (struct sockaddr_in *)if_addrs->ifa_addr;
struct sockaddr_in *ifu_netmask = (struct sockaddr_in *)if_addrs->ifa_netmask;
if (ifu_localaddr->sin_addr.s_addr != htonl(INADDR_ANY)) {
if ((ifu_localaddr->sin_addr.s_addr & ifu_netmask->sin_addr.s_addr) ==
(client_addr.sin_addr.s_addr & ifu_netmask->sin_addr.s_addr)) {
local_ip = ifu_localaddr->sin_addr.s_addr;
found = true;
break;
}
}
}
if_addrs = if_addrs->ifa_next;
}
if (addrs) {
freeifaddrs(addrs);
}
return found;
}
size_t RecvFrom(socket_t &sock, void *buff, size_t buf_size, int flag, struct sockaddr *addr, int *addrlen) {
return recvfrom(sock, buff, buf_size, 0, addr, (socklen_t *)addrlen);
}
} // namespace util
} // namespace livox
#endif // WIN32
sdk_core/src/base/network_util.cpp → sdk_core/src/base/network/win/network_util.cpp
+46 -72
View File
@@ -22,65 +22,64 @@
// SOFTWARE.
//
#include "network_util.h"
#ifdef WIN32
#include "base/network/network_util.h"
#include <memory>
#include <Winsock2.h>
#include <iphlpapi.h>
#include <string>
#pragma comment(lib,"iphlpapi.lib")
#pragma comment(lib,"ws2_32.lib")
#else
#include <ifaddrs.h>
#endif
namespace livox {
namespace util {
apr_socket_t *CreateBindSocket(uint16_t port, apr_pool_t *mem_pool, bool reuse_port, bool nonblock) {
apr_sockaddr_t *sa = NULL;
apr_socket_t *s = NULL;
apr_status_t rv = APR_SUCCESS;
void CloseSock(socket_t sock) {
closesocket(sock);
}
rv = apr_sockaddr_info_get(&sa, NULL, APR_INET, port, 0, mem_pool);
if (rv != APR_SUCCESS) {
return NULL;
}
rv = apr_socket_create(&s, sa->family, SOCK_DGRAM, APR_PROTO_UDP, mem_pool);
if (rv != APR_SUCCESS) {
return NULL;
}
if (reuse_port) {
rv = apr_socket_opt_set(s, APR_SO_REUSEADDR, 1);
if (rv != APR_SUCCESS) {
apr_socket_close(s);
s = NULL;
return NULL;
}
}
rv = apr_socket_bind(s, sa);
if (rv != APR_SUCCESS) {
apr_socket_close(s);
s = NULL;
return NULL;
socket_t CreateSocket(uint16_t port, bool nonblock, bool reuse_port) {
int status = -1;
int on = -1;
int sock = -1;
struct sockaddr_in servaddr;
sock = socket(AF_INET, SOCK_DGRAM, 0);
if (sock == INVALID_SOCKET) {
return -1;
}
if (nonblock) {
apr_socket_opt_set(s, APR_SO_NONBLOCK, 1);
apr_socket_timeout_set(s, 0);
status = ioctlsocket(sock, FIONBIO, (u_long *)&on);
if (status != NO_ERROR) {
closesocket(sock);
return -1;
}
}
return s;
if (reuse_port) {
status = setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
(char *) &on, sizeof (on));
if (status != 0) {
closesocket(sock);
return -1;
}
}
memset(&servaddr, 0, sizeof(servaddr));
// Filling server information
servaddr.sin_family = AF_INET; // IPv4
servaddr.sin_addr.s_addr = INADDR_ANY;
servaddr.sin_port = htons(port);
status = bind(sock, (const struct sockaddr *)&servaddr, sizeof(servaddr));
if (status != 0) {
closesocket(sock);
return -1;
}
return sock;
}
#ifdef WIN32
bool GetAdapterState(const IP_ADAPTER_INFO *pAdapter)
{
bool GetAdapterState(const IP_ADAPTER_INFO *pAdapter) {
if(pAdapter == NULL) {
return false;
}
@@ -129,37 +128,12 @@ bool FindLocalIp(const struct sockaddr_in &client_addr, uint32_t &local_ip) {
}
return found;
}
#else
bool FindLocalIp(const struct sockaddr_in &client_addr, uint32_t &local_ip) {
struct ifaddrs *if_addrs = NULL, *addrs = NULL;
if (getifaddrs(&if_addrs) == -1) {
return false;
}
addrs = if_addrs;
bool found = false;
while (if_addrs != NULL) {
if ((if_addrs->ifa_addr != NULL) && (if_addrs->ifa_addr->sa_family == AF_INET)) // check it is IP4
{
// is a connected IP4 Address
struct sockaddr_in *ifu_localaddr = (struct sockaddr_in *)if_addrs->ifa_addr;
struct sockaddr_in *ifu_netmask = (struct sockaddr_in *)if_addrs->ifa_netmask;
if (ifu_localaddr->sin_addr.s_addr != htonl(INADDR_ANY)) {
if ((ifu_localaddr->sin_addr.s_addr & ifu_netmask->sin_addr.s_addr) ==
(client_addr.sin_addr.s_addr & ifu_netmask->sin_addr.s_addr)) {
local_ip = ifu_localaddr->sin_addr.s_addr;
found = true;
break;
}
}
}
if_addrs = if_addrs->ifa_next;
}
if (addrs) {
freeifaddrs(addrs);
}
return found;
size_t RecvFrom(socket_t &sock, void *buff, size_t buf_size, int flag, struct sockaddr *addr, int* addrlen) {
return recvfrom(sock, (char *)buff, buf_size, 0, addr, addrlen);
}
#endif
} // namespace util
} // namespace livox
#endif // WIN32
sdk_core/src/base/thread_base.cpp
+9 -28
View File
@@ -23,49 +23,30 @@
//
#include "thread_base.h"
#include <thread>
namespace livox {
static void *APR_THREAD_FUNC ClassThreadHelperFunc(apr_thread_t *thd, void *data) {
ThreadBase *caller = static_cast<ThreadBase *>(data);
if (caller == NULL) {
return NULL;
}
caller->ThreadFunc();
apr_thread_exit(thd, APR_SUCCESS);
return NULL;
}
ThreadBase::ThreadBase() : thread_(NULL), quit_(false), pool_(NULL) {}
ThreadBase::ThreadBase() : quit_(false), is_thread_valid_(false) {}
bool ThreadBase::Start() {
quit_ = false;
apr_threadattr_t *thread_attr = NULL;
apr_status_t rv = APR_SUCCESS;
rv = apr_threadattr_create(&thread_attr, pool_);
if (rv != APR_SUCCESS) {
return false;
}
rv = apr_thread_create(&thread_, thread_attr, ClassThreadHelperFunc, this, pool_);
return rv == APR_SUCCESS;
thread_ = std::make_shared<std::thread>(&ThreadBase::ThreadFunc, this);
is_thread_valid_ = true;
return true;
}
void ThreadBase::Join() {
apr_status_t rv = APR_SUCCESS;
if (thread_) {
apr_thread_join(&rv, thread_);
thread_ = NULL;
if (is_thread_valid_) {
thread_->join();
}
}
}
bool ThreadBase::Init() {
return apr_pool_create(&pool_, NULL) == APR_SUCCESS;
return true;
}
void ThreadBase::Uninit() {
if (pool_) {
apr_pool_destroy(pool_);
pool_ = NULL;
}
}
} // namespace livox
sdk_core/src/base/thread_base.h
+5 -5
View File
@@ -24,10 +24,8 @@
#ifndef LIVOX_THREAD_BASE_H_
#define LIVOX_THREAD_BASE_H_
#include <apr_general.h>
#include <apr_thread_proc.h>
#include <apr_portable.h>
#include <atomic>
#include <thread>
#include "noncopyable.h"
namespace livox {
@@ -45,9 +43,11 @@ class ThreadBase : public noncopyable {
bool IsQuit() { return quit_; }
protected:
apr_thread_t *thread_;
std::shared_ptr<std::thread> thread_;
std::atomic_bool quit_;
apr_pool_t *pool_;
private:
std::atomic_bool is_thread_valid_;
};
} // namespace livox
sdk_core/src/base/wake_up/unix/wake_up_pipe.cpp
+103
View File
@@ -0,0 +1,103 @@
//
// The MIT License (MIT)
//
// Copyright (c) 2019 Livox. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
#ifndef WIN32
#include "base/wake_up/wake_up_pipe.h"
#include <fcntl.h>
#include <unistd.h>
#include <stdio.h>
namespace livox {
WakeUpPipe::~WakeUpPipe() {
PipeDestroy();
}
bool WakeUpPipe::WakeUp() {
char ch = '1';
ssize_t nbytes = sizeof(ch);
if (pipe_in_ > 0) {
if (nbytes != write(pipe_in_, &ch, nbytes)) {
return false;
}
}
return true;
}
bool WakeUpPipe::Drain() {
char ch[512];
size_t size = sizeof(ch);
if (pipe_out_ > 0) {
if (read(pipe_out_, ch, size) < 0) {
return false;
}
}
return true;
}
bool WakeUpPipe::PipeDestroy() {
if (pipe_in_ > 0) {
close(pipe_in_);
}
if (pipe_out_ > 0) {
close(pipe_out_);
}
return true;
}
bool WakeUpPipe::PipeCreate() {
//in filedes[0]
//out filedes[1]
int filedes[2];
if (pipe(filedes) == -1) {
return false;
}
int flags = 0;
if ((flags = fcntl(filedes[0], F_GETFL|O_NONBLOCK)) == -1) {
return false;
}
flags |= FD_CLOEXEC;
if (fcntl(filedes[0], F_SETFL, flags) == -1) {
return false;
}
flags = 0;
if ((flags = fcntl(filedes[1], F_GETFD)) == -1) {
return false;
}
flags |= FD_CLOEXEC;
if (fcntl(filedes[1], F_SETFD, flags) == -1) {
return false;
}
pipe_out_ = filedes[0];
pipe_in_ = filedes[1];
return true;
}
} // namespace livox
#endif // WIN32
sdk_core/src/base/util.h → sdk_core/src/base/wake_up/wake_up_pipe.h
+17 -9
View File
@@ -22,16 +22,24 @@
// SOFTWARE.
//
#ifndef LIVOX_UTIL_H_
#define LIVOX_UTIL_H_
#include <string>
#include "apr_general.h"
#include "apr_time.h"
#ifndef WAKE_UP_PIPE_H_
#define WAKE_UP_PIPE_H_
namespace livox {
class WakeUpPipe {
public:
WakeUpPipe(): pipe_in_(0), pipe_out_(0) {}
virtual ~WakeUpPipe();
bool PipeCreate();
bool PipeDestroy();
bool WakeUp();
bool Drain();
int GetPipeOut() { return pipe_out_; }
protected:
int pipe_in_;
int pipe_out_;
};
std::string PrintAPRStatus(apr_status_t s);
std::string PrintAPRTime(apr_time_t t);
} // namespace livox
} // namespace livox
#endif // LIVOX_UTIL_H_
#endif // WAKE_UP_PIPE_H_
sdk_core/src/base/wake_up/win/wake_up_pipe.cpp
+132
View File
@@ -0,0 +1,132 @@
//
// The MIT License (MIT)
//
// Copyright (c) 2019 Livox. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
#ifdef WIN32
#include "base/wake_up/wake_up_pipe.h"
#include <fcntl.h>
#include <winsock2.h>
#include <stdio.h>
#pragma comment(lib,"iphlpapi.lib")
#pragma comment(lib,"ws2_32.lib")
namespace livox {
WakeUpPipe::~WakeUpPipe() {
PipeDestroy();
}
bool WakeUpPipe::WakeUp() {
char ch = '1';
size_t nbytes = sizeof(ch);
if (pipe_in_ > 0) {
if (nbytes != send(pipe_in_,&ch, nbytes, 0)) {
return false;
}
}
return true;
}
bool WakeUpPipe::Drain() {
char ch[512];
size_t size = sizeof(ch);
if (pipe_out_ > 0) {
recv(pipe_out_, ch, size, 0);
}
return true;
}
bool WakeUpPipe::PipeDestroy() {
if (pipe_in_ > 0) {
closesocket(pipe_in_);
}
if (pipe_out_ > 0) {
closesocket(pipe_out_);
}
return true;
}
bool WakeUpPipe::PipeCreate() {
int listen_sock = -1;
unsigned long on = 1;
bool status = false;
if ((listen_sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) == INVALID_SOCKET) {
return false;
}
struct sockaddr_in servaddr;
int servaddr_len = sizeof(servaddr);
servaddr.sin_family = AF_INET;
servaddr.sin_port = 0;
servaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
do {
if (bind(listen_sock, (const struct sockaddr *)&servaddr, sizeof(servaddr)) == SOCKET_ERROR) {
break;
}
if (getsockname(listen_sock, (struct sockaddr *)&servaddr, &servaddr_len) == SOCKET_ERROR) {
break;
}
if (listen(listen_sock, 1) == SOCKET_ERROR) {
break;
}
if ((pipe_in_ = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) == INVALID_SOCKET) {
break;
}
if (connect(pipe_in_, (const struct sockaddr *)&servaddr, sizeof(servaddr)) == SOCKET_ERROR) {
break;
}
if (ioctlsocket(listen_sock, FIONBIO, &on) == SOCKET_ERROR) {
break;
}
struct sockaddr_in clientaddr;
int clientaddr_len = sizeof(clientaddr);
fd_set poll_set;
FD_ZERO(&poll_set);
FD_SET(listen_sock, &poll_set);
// timeout 2s
struct timeval timeout = {2, 0};
int rv = select(0, &poll_set, nullptr, nullptr, &timeout);
if (rv <= 0) {
break;
}
if ((pipe_out_ = accept(listen_sock, (struct sockaddr *)&clientaddr, &clientaddr_len)) == INVALID_SOCKET) {
break;
}
status = true;
} while(0);
closesocket(listen_sock);
if (!status) {
if (pipe_in_ > 0) {
closesocket(pipe_in_);
}
return false;
}
return true;
}
} // namespace livox
#endif // WIN32
sdk_core/src/command_handler/command_channel.cpp
+48 -48
View File
@@ -26,10 +26,11 @@
#include <functional>
#include <atomic>
#include "base/logging.h"
#include "base/network_util.h"
#include "base/network/network_util.h"
#include "command_impl.h"
#include "device_manager.h"
#include "livox_def.h"
#include <stdio.h>
using std::bind;
using std::list;
@@ -41,49 +42,46 @@ using std::chrono::steady_clock;
namespace livox {
CommandChannel::CommandChannel(apr_port_t port,
CommandChannel::CommandChannel(uint16_t port,
uint8_t handle,
const string &remote_ip,
CommandChannelDelegate *cb,
apr_pool_t *pool)
CommandChannelDelegate *cb)
: handle_(handle),
port_(port),
sock_(NULL),
mem_pool_(pool),
loop_(NULL),
loop_(),
callback_(cb),
comm_port_(new CommPort),
heartbeat_time_(),
remote_ip_(remote_ip),
heartbeat_time_(),
last_heartbeat_() {}
bool CommandChannel::Bind(IOLoop *loop) {
if (loop == NULL) {
bool CommandChannel::Bind(std::weak_ptr<IOLoop> loop) {
if (loop.expired()) {
return false;
}
loop_ = loop;
sock_ = util::CreateBindSocket(port_, mem_pool_);
if (sock_ == NULL) {
sock_ = util::CreateSocket(port_);
if (sock_ == -1) {
return false;
}
loop_->AddDelegate(sock_, this);
loop_.lock()->AddDelegate(sock_, this);
last_heartbeat_ = steady_clock::now();
return true;
}
void CommandChannel::OnData(apr_socket_t *, void *) {
apr_sockaddr_t addr;
void CommandChannel::OnData(socket_t , void *) {
struct sockaddr addr;
int addrlen = sizeof(addr);
uint32_t buf_size = 0;
uint8_t *cache_buf = comm_port_->FetchCacheFreeSpace(&buf_size);
apr_size_t size = buf_size;
apr_status_t rv = apr_socket_recvfrom(&addr, sock_, 0, reinterpret_cast<char *>(cache_buf), &size);
comm_port_->UpdateCacheWrIdx(size);
if (rv != APR_SUCCESS) {
LOG_ERROR(PrintAPRStatus(rv));
int size = buf_size;
size = util::RecvFrom(sock_, reinterpret_cast<char *>(cache_buf), buf_size, 0, &addr, &addrlen);
if (size <= 0) {
return;
}
comm_port_->UpdateCacheWrIdx(size);
CommPacket packet;
memset(&packet, 0, sizeof(packet));
@@ -115,8 +113,14 @@ void CommandChannel::OnData(apr_socket_t *, void *) {
void CommandChannel::SendAsync(const Command &command) {
Command cmd = DeepCopy(command);
if (loop_) {
loop_->PostTask(bind(&CommandChannel::Send, this, cmd));
if (!loop_.expired()) {
auto w_ptr = WeakProtector(protector_);
loop_.lock()->PostTask([this, w_ptr, cmd](){
if(w_ptr.expired()) {
return;
}
Send(cmd);
});
}
}
@@ -134,7 +138,7 @@ void CommandChannel::OnTimer(TimePoint now) {
}
for (list<Command>::iterator ite = timeout_commands.begin(); ite != timeout_commands.end(); ++ite) {
LOG_WARN("Command Timeout: Set {}, Id {}, Seq {}",
LOG_WARN("Command Timeout: Set {}, Id {}, Seq {}",
(uint16_t)ite->packet.cmd_set, ite->packet.cmd_code, ite->packet.seq_num);
if (callback_) {
ite->packet.packet_type = kCommandTypeAck;
@@ -150,16 +154,12 @@ void CommandChannel::OnTimer(TimePoint now) {
}
void CommandChannel::Uninit() {
if (sock_) {
apr_os_thread_t thread_id = apr_os_thread_current();
if (apr_os_thread_equal(loop_->GetThreadId(), thread_id)) {
loop_->RemoveDelegateSync(sock_);
} else {
loop_->RemoveDelegate(sock_, this);
if (sock_ != -1) {
if (!loop_.expired()) {
loop_.lock()->RemoveDelegate(sock_, this);
}
loop_ = NULL;
apr_socket_close(sock_);
sock_ = NULL;
util::CloseSock(sock_);
sock_ = -1;
}
callback_ = NULL;
if (comm_port_) {
@@ -190,22 +190,22 @@ void CommandChannel::HeartBeat(TimePoint t) {
}
void CommandChannel::SendInternal(const Command &command) {
apr_pool_t *subpool = NULL;
apr_pool_create(&subpool, mem_pool_);
uint8_t *buf = (uint8_t *)apr_palloc(subpool, kMaxCommandBufferSize);
uint32_t size = 0;
comm_port_->Pack(buf, kMaxCommandBufferSize, &size, command.packet);
apr_size_t apr_size = size;
apr_status_t rv = APR_SUCCESS;
apr_sockaddr_t *sa = NULL;
rv = apr_sockaddr_info_get(&sa, remote_ip_.c_str(), APR_INET, 65000, 0, subpool);
if (rv == APR_SUCCESS) {
rv = apr_socket_sendto(sock_, sa, 0, (const char *)buf, &apr_size);
std::vector<uint8_t> buf(kMaxCommandBufferSize + 1);
int size = 0;
comm_port_->Pack(buf.data(), kMaxCommandBufferSize, (uint32_t *)&size, command.packet);
struct sockaddr_in servaddr;
servaddr.sin_family = AF_INET;
servaddr.sin_port = htons(65000);
servaddr.sin_addr.s_addr = inet_addr(remote_ip_.c_str());
int byte_send = sendto(sock_, (const char*)buf.data(), size, 0, (const struct sockaddr *) &servaddr,
sizeof(servaddr));
if (byte_send < 0) {
if (command.cb) {
(*command.cb)(kStatusSendFailed, handle_, NULL);
}
}
if (rv != APR_SUCCESS) {
(*command.cb)(kStatusSendFailed, handle_, NULL);
}
apr_pool_destroy(subpool);
}
uint16_t CommandChannel::GenerateSeq() {
sdk_core/src/command_handler/command_channel.h
+15 -13
View File
@@ -28,8 +28,7 @@
#include <map>
#include <list>
#include <string>
#include <chrono>
#include "apr_network_io.h"
#include <algorithm>
#include "base/io_loop.h"
#include "comm/comm_port.h"
@@ -67,18 +66,19 @@ class CommandChannelDelegate {
virtual void OnHeartbeatStateUpdate(uint8_t handle, const HeartbeatResponse &state) = 0;
};
class Protector {};
/**
* CommandChannel implements the sending/receiving commands with a specific device.
*/
class CommandChannel : public IOLoop::IOLoopDelegate {
public:
typedef std::chrono::steady_clock::time_point TimePoint;
CommandChannel(apr_port_t port,
CommandChannel(uint16_t port,
uint8_t handle,
const std::string &remote_ip,
CommandChannelDelegate *cb,
apr_pool_t *pool);
CommandChannelDelegate *cb);
virtual ~CommandChannel() { Uninit(); }
/** Uninitialize CommandChannel. */
@@ -89,7 +89,7 @@ class CommandChannel : public IOLoop::IOLoopDelegate {
* @param loop the IOLoop to bind.
* @return true on successfully.
*/
bool Bind(IOLoop *loop);
bool Bind(std::weak_ptr<IOLoop> loop);
/**
* Send a command asynchronously.
@@ -97,7 +97,7 @@ class CommandChannel : public IOLoop::IOLoopDelegate {
*/
void SendAsync(const Command &command);
void OnData(apr_socket_t *, void *);
void OnData(socket_t, void *);
void OnTimer(TimePoint now);
static uint16_t GenerateSeq();
@@ -113,16 +113,18 @@ class CommandChannel : public IOLoop::IOLoopDelegate {
private:
const std::chrono::milliseconds kHeartbeatTimer = std::chrono::milliseconds(800);
uint8_t handle_;
apr_port_t port_;
apr_socket_t *sock_;
apr_pool_t *mem_pool_;
IOLoop *loop_;
uint16_t port_;
socket_t sock_ = -1;
std::weak_ptr<IOLoop> loop_;
CommandChannelDelegate *callback_;
std::map<uint16_t, std::pair<Command, TimePoint> > commands_;
std::unique_ptr<CommPort> comm_port_;
TimePoint heartbeat_time_;
std::string remote_ip_;
TimePoint heartbeat_time_;
TimePoint last_heartbeat_;
using SharedProtecotr = std::shared_ptr<Protector>;
using WeakProtector = std::weak_ptr<Protector>;
SharedProtecotr protector_ = std::make_shared<Protector>();
};
} // namespace livox
sdk_core/src/command_handler/command_handler.cpp
+3 -14
View File
@@ -90,9 +90,9 @@ bool CommandHandler::AddDevice(const DeviceInfo &info) {
if (impl_ == NULL) {
DeviceMode mode = static_cast<DeviceMode>(device_manager().device_mode());
if (mode == kDeviceModeHub) {
impl_.reset(new HubCommandHandlerImpl(this, mem_pool_, loop_));
impl_.reset(new HubCommandHandlerImpl(this, loop_));
} else if (mode == kDeviceModeLidar) {
impl_.reset(new LidarCommandHandlerImpl(this, mem_pool_, loop_));
impl_.reset(new LidarCommandHandlerImpl(this, loop_));
}
}
if (impl_ == NULL) {
@@ -102,26 +102,15 @@ bool CommandHandler::AddDevice(const DeviceInfo &info) {
return impl_->AddDevice(info);
}
bool CommandHandler::Init(IOLoop *loop) {
apr_status_t rv = apr_pool_create(&mem_pool_, NULL);
if (rv != APR_SUCCESS) {
return false;
}
bool CommandHandler::Init(std::weak_ptr<IOLoop> loop) {
loop_ = loop;
return true;
}
void CommandHandler::Uninit() {
loop_ = NULL;
if (impl_) {
impl_.reset(NULL);
}
if (mem_pool_) {
apr_pool_destroy(mem_pool_);
mem_pool_ = NULL;
}
}
void CommandHandler::OnCommand(uint8_t handle, const Command &command) {
sdk_core/src/command_handler/command_handler.h
+3 -5
View File
@@ -29,7 +29,6 @@
#include <map>
#include <mutex>
#include "base/command_callback.h"
#include "base/util.h"
#include "command_channel.h"
#include "device_discovery.h"
#include "livox_sdk.h"
@@ -39,9 +38,9 @@ class CommandHandlerImpl;
class CommandHandler : public noncopyable {
public:
explicit CommandHandler() : mem_pool_(NULL), loop_(NULL) {}
explicit CommandHandler() {}
bool Init(IOLoop *loop);
bool Init(std::weak_ptr<IOLoop> loop);
void Uninit();
bool AddDevice(const DeviceInfo &info);
@@ -71,10 +70,9 @@ class CommandHandler : public noncopyable {
private:
std::multimap<uint16_t, Command> message_registers_;
apr_pool_t *mem_pool_;
std::unique_ptr<CommandHandlerImpl> impl_;
std::mutex mutex_;
IOLoop *loop_;
std::weak_ptr<IOLoop> loop_;
};
class CommandHandlerImpl : public CommandChannelDelegate {
sdk_core/src/command_handler/command_impl.cpp
+1 -1
View File
@@ -262,7 +262,7 @@ livox_status HubStopSampling(CommonCommandCallback cb, void *client_data) {
return DeviceSampleControl(kHubDefaultHandle, false, cb, client_data);
}
livox_status HubGetLidarHandle(uint8_t slot, uint8_t id) {
uint8_t HubGetLidarHandle(uint8_t slot, uint8_t id) {
return (slot - 1) * 3 + id - 1;
}
sdk_core/src/command_handler/hub_command_handler.cpp
+2 -2
View File
@@ -23,7 +23,7 @@
//
#include "hub_command_handler.h"
#include "base/network_util.h"
#include "base/network/network_util.h"
namespace livox {
@@ -41,7 +41,7 @@ bool HubCommandHandlerImpl::AddDevice(const DeviceInfo &info) {
is_valid_ = true;
hub_info_ = info;
channel_.reset(new CommandChannel(info.cmd_port, info.handle, info.ip, this, mem_pool_));
channel_.reset(new CommandChannel(info.cmd_port, info.handle, info.ip, this));
channel_->Bind(loop_);
return true;
}
sdk_core/src/command_handler/hub_command_handler.h
+3 -4
View File
@@ -31,16 +31,15 @@
namespace livox {
class HubCommandHandlerImpl : public CommandHandlerImpl {
public:
HubCommandHandlerImpl(CommandHandler *handler, apr_pool_t *pool, IOLoop *loop)
: CommandHandlerImpl(handler), mem_pool_(pool), loop_(loop), is_valid_(false) {}
HubCommandHandlerImpl(CommandHandler *handler, std::weak_ptr<IOLoop> loop)
: CommandHandlerImpl(handler), loop_(loop), is_valid_(false) {}
void Uninit();
bool AddDevice(const DeviceInfo &info);
bool RemoveDevice(uint8_t handle);
livox_status SendCommand(uint8_t handle, const Command &command);
private:
apr_pool_t *mem_pool_;
IOLoop *loop_;
std::weak_ptr<IOLoop> loop_;
bool is_valid_;
DeviceInfo hub_info_;
std::unique_ptr<CommandChannel> channel_;
sdk_core/src/command_handler/lidar_command_handler.cpp
+1 -1
View File
@@ -34,7 +34,7 @@ void LidarCommandHandlerImpl::Uninit() {
bool LidarCommandHandlerImpl::AddDevice(const DeviceInfo &info) {
std::shared_ptr<CommandChannel> channel =
std::make_shared<CommandChannel>(info.cmd_port, info.handle, info.ip, this, mem_pool_);
std::make_shared<CommandChannel>(info.cmd_port, info.handle, info.ip, this);
channel->Bind(loop_);
DeviceItem item = {channel, info};
sdk_core/src/command_handler/lidar_command_handler.h
+3 -4
View File
@@ -31,8 +31,8 @@ namespace livox {
class LidarCommandHandlerImpl : public CommandHandlerImpl {
public:
LidarCommandHandlerImpl(CommandHandler *handler, apr_pool_t *pool, IOLoop *loop)
: CommandHandlerImpl(handler), mem_pool_(pool), loop_(loop) {}
LidarCommandHandlerImpl(CommandHandler *handler, std::weak_ptr<IOLoop> loop)
: CommandHandlerImpl(handler), loop_(loop) {}
void Uninit();
@@ -45,9 +45,8 @@ class LidarCommandHandlerImpl : public CommandHandlerImpl {
std::shared_ptr<CommandChannel> channel;
DeviceInfo info;
} DeviceItem;
apr_pool_t *mem_pool_;
std::list<DeviceItem> devices_;
IOLoop *loop_;
std::weak_ptr<IOLoop> loop_;
};
} // namespace livox
sdk_core/src/data_handler/data_handler.cpp
+2 -10
View File
@@ -49,9 +49,9 @@ bool DataHandler::AddDevice(const DeviceInfo &info) {
if (impl_ == NULL) {
DeviceMode mode = static_cast<DeviceMode>(device_manager().device_mode());
if (mode == kDeviceModeHub) {
impl_.reset(new HubDataHandlerImpl(this, mem_pool_));
impl_.reset(new HubDataHandlerImpl(this));
} else if (mode == kDeviceModeLidar) {
impl_.reset(new LidarDataHandlerImpl(this, mem_pool_));
impl_.reset(new LidarDataHandlerImpl(this));
}
if (impl_ == NULL || !impl_->Init()) {
@@ -63,10 +63,6 @@ bool DataHandler::AddDevice(const DeviceInfo &info) {
}
bool DataHandler::Init() {
apr_status_t rv = apr_pool_create(&mem_pool_, NULL);
if (rv != APR_SUCCESS) {
return false;
}
return true;
}
@@ -74,10 +70,6 @@ void DataHandler::Uninit() {
if (impl_) {
impl_.reset(NULL);
}
if (mem_pool_) {
apr_pool_destroy(mem_pool_);
mem_pool_ = NULL;
}
}
void DataHandler::OnDataCallback(uint8_t handle, void *data, uint16_t size) {
sdk_core/src/data_handler/data_handler.h
+1 -3
View File
@@ -29,7 +29,6 @@
#include <functional>
#include <memory>
#include <mutex>
#include "apr_pools.h"
#include "base/io_thread.h"
#include "device_manager.h"
@@ -41,7 +40,7 @@ class DataHandler : public noncopyable {
typedef std::function<void(uint8_t handle, LivoxEthPacket *data, uint32_t data_num, void *client_data)> DataCallback;
public:
DataHandler() : mem_pool_(NULL) {}
DataHandler() {}
bool Init();
void Uninit();
@@ -53,7 +52,6 @@ class DataHandler : public noncopyable {
void OnDataCallback(uint8_t handle, void *data, uint16_t size);
private:
apr_pool_t *mem_pool_;
std::array<DataCallback, kMaxConnectedDeviceNum> callbacks_;
std::array<void *, kMaxConnectedDeviceNum> client_data_;
std::unique_ptr<DataHandlerImpl> impl_;
sdk_core/src/data_handler/hub_data_handler.cpp
+31 -16
View File
@@ -24,7 +24,14 @@
#include "hub_data_handler.h"
#include <base/logging.h>
#include "base/network_util.h"
#include "base/network/network_util.h"
#ifdef WIN32
#include <winsock2.h>
#include <ws2tcpip.h>
#include <ws2def.h>
#else
#include "arpa/inet.h"
#endif
namespace livox {
@@ -43,12 +50,11 @@ void HubDataHandlerImpl::Uninit() {
thread_.reset(NULL);
}
if (sock_) {
apr_socket_close(sock_);
sock_ = NULL;
if (sock_ > 0) {
util::CloseSock(sock_);
sock_ = -1;
}
is_valid_ = false;
mem_pool_ = NULL;
}
bool HubDataHandlerImpl::AddDevice(const DeviceInfo &info) {
@@ -58,23 +64,29 @@ bool HubDataHandlerImpl::AddDevice(const DeviceInfo &info) {
is_valid_ = true;
hub_info_ = info;
sock_ = util::CreateBindSocket(info.data_port, mem_pool_);
if (sock_ == NULL) {
sock_ = util::CreateSocket(info.data_port);
if (sock_ < 0) {
is_valid_ = false;
return false;
}
thread_->loop()->AddDelegate(sock_, this);
auto loop = thread_->loop();
if (!loop.expired()) {
loop.lock()->AddDelegate(sock_, this);
}
return true;
}
void HubDataHandlerImpl::OnData(apr_socket_t *, void *) {
apr_sockaddr_t addr;
void HubDataHandlerImpl::OnData(socket_t, void *) {
struct sockaddr addr;
int addrlen = sizeof(addr);
if (buf_.size() < kMaxBufferSize) {
buf_.resize(kMaxBufferSize);
}
apr_size_t size = kMaxBufferSize;
apr_socket_recvfrom(&addr, sock_, 0, &buf_[0], &size);
int size = kMaxBufferSize;
size = util::RecvFrom(sock_, reinterpret_cast<char *>(&buf_[0]), kMaxBufferSize, 0, &addr, &addrlen);
if (size <= 0) {
return;
}
if (handler_) {
handler_->OnDataCallback(hub_info_.handle, &buf_[0], size);
@@ -83,9 +95,12 @@ void HubDataHandlerImpl::OnData(apr_socket_t *, void *) {
void HubDataHandlerImpl::RemoveDevice(uint8_t handle) {
if (handle == hub_info_.handle) {
thread_->loop()->RemoveDelegate(sock_, this);
apr_socket_close(sock_);
sock_ = NULL;
auto loop = thread_->loop();
if (!loop.expired()) {
loop.lock()->RemoveDelegate(sock_, this);
}
util::CloseSock(sock_);
sock_ = -1;
is_valid_ = false;
}
}
sdk_core/src/data_handler/hub_data_handler.h
+4 -5
View File
@@ -32,8 +32,8 @@
namespace livox {
class HubDataHandlerImpl : public DataHandlerImpl {
public:
HubDataHandlerImpl(DataHandler *handler, apr_pool_t *mem_pool)
: DataHandlerImpl(handler), mem_pool_(mem_pool), thread_(new IOThread()), sock_(NULL), is_valid_(false) {}
HubDataHandlerImpl(DataHandler *handler)
: DataHandlerImpl(handler), thread_(new IOThread()), is_valid_(false) {}
~HubDataHandlerImpl() { Uninit(); }
bool Init();
@@ -41,12 +41,11 @@ class HubDataHandlerImpl : public DataHandlerImpl {
bool AddDevice(const DeviceInfo &info);
void RemoveDevice(uint8_t t);
void OnData(apr_socket_t *sock, void *client_data);
void OnData(socket_t sock, void *client_data);
private:
apr_pool_t *mem_pool_;
std::unique_ptr<IOThread> thread_;
apr_socket_t *sock_;
socket_t sock_ = -1;
DeviceInfo hub_info_;
bool is_valid_;
std::vector<char> buf_;
sdk_core/src/data_handler/lidar_data_handler.cpp
+29 -22
View File
@@ -24,7 +24,15 @@
#include "lidar_data_handler.h"
#include <mutex>
#include "base/network_util.h"
#include "base/network/network_util.h"
#ifdef WIN32
#include <winsock2.h>
#include <ws2def.h>
#pragma comment(lib, "Ws2_32.lib")
#else
#include <unistd.h>
#endif // WIN32
using std::lock_guard;
using std::mutex;
@@ -41,16 +49,14 @@ void LidarDataHandlerImpl::Uninit() {
for (list<DeviceItem>::iterator ite = devices_.begin(); ite != devices_.end(); ++ite) {
DeviceItem &item = *ite;
if (item.thread) {
if (item.thread->loop()) {
item.thread->loop()->RemoveDelegate(item.sock, this);
}
item.thread->loop().lock()->RemoveDelegate(item.sock, this);
item.thread->Quit();
item.thread->Join();
item.thread->Uninit();
}
if (item.sock) {
apr_socket_close(item.sock);
if (item.sock > 0) {
util::CloseSock(item.sock);
}
}
@@ -58,14 +64,11 @@ void LidarDataHandlerImpl::Uninit() {
}
bool LidarDataHandlerImpl::AddDevice(const DeviceInfo &info) {
apr_socket_t *sock = util::CreateBindSocket(info.data_port, mem_pool_);
if (sock == NULL) {
return false;
}
socket_t sock = util::CreateSocket(info.data_port);
shared_ptr<IOThread> thread = std::make_shared<IOThread>();
std::shared_ptr<IOThread> thread = std::make_shared<IOThread>();
thread->Init(false, false);
thread->loop()->AddDelegate(sock, this, reinterpret_cast<void *>(info.handle));
thread->loop().lock()->AddDelegate(sock, this, reinterpret_cast<void *>(info.handle));
{
lock_guard<mutex> lock(mutex_);
DeviceItem item = {sock, thread, info.handle};
@@ -90,17 +93,19 @@ void LidarDataHandlerImpl::RemoveDevice(uint8_t handle) {
}
if (found) {
item.thread->loop()->RemoveDelegate(item.sock, this);
item.thread->loop().lock()->RemoveDelegate(item.sock, this);
item.thread->Quit();
item.thread->Join();
item.thread->Uninit();
if (item.sock) {
apr_socket_close(item.sock);
if (item.sock > 0) {
util::CloseSock(item.sock);
}
}
}
void LidarDataHandlerImpl::OnData(apr_socket_t *sock, void *client_data) {
void LidarDataHandlerImpl::OnData(socket_t sock, void *client_data) {
struct sockaddr addr;
int addrlen = sizeof(addr);
uint8_t handle = static_cast<uint8_t>(reinterpret_cast<uintptr_t>(client_data));
if (handle > data_buffers_.size()) {
return;
@@ -110,13 +115,15 @@ void LidarDataHandlerImpl::OnData(apr_socket_t *sock, void *client_data) {
buf.reset(new char[kMaxBufferSize]);
}
apr_sockaddr_t addr;
apr_size_t size = kMaxBufferSize;
if (APR_SUCCESS == apr_socket_recvfrom(&addr, sock, 0, buf.get(), &size)) {
if (handler_) {
handler_->OnDataCallback(handle, buf.get(), size);
}
int size = kMaxBufferSize;
size = util::RecvFrom(sock, reinterpret_cast<char *>(buf.get()), kMaxBufferSize, 0, &addr, &addrlen);
if (size <= 0) {
return;
}
if (handler_) {
handler_->OnDataCallback(handle, buf.get(), size);
}
}
} // namespace livox
sdk_core/src/data_handler/lidar_data_handler.h
+3 -4
View File
@@ -34,24 +34,23 @@ namespace livox {
class LidarDataHandlerImpl : public DataHandlerImpl {
public:
LidarDataHandlerImpl(DataHandler *handler, apr_pool_t *mem_pool) : DataHandlerImpl(handler), mem_pool_(mem_pool) {}
LidarDataHandlerImpl(DataHandler *handler) : DataHandlerImpl(handler) {}
~LidarDataHandlerImpl() { Uninit(); }
bool Init();
void Uninit();
bool AddDevice(const DeviceInfo &info);
void RemoveDevice(uint8_t handle);
void OnData(apr_socket_t *sock, void *client_data);
void OnData(socket_t sock, void *client_data);
private:
typedef struct {
apr_socket_t *sock;
socket_t sock;
std::shared_ptr<IOThread> thread;
uint16_t handle;
} DeviceItem;
std::list<DeviceItem> devices_;
std::array<std::unique_ptr<char[]>, kMaxConnectedDeviceNum> data_buffers_;
apr_pool_t *mem_pool_;
std::mutex mutex_;
};
sdk_core/src/device_discovery.cpp
+51 -75
View File
@@ -27,15 +27,8 @@
#include <mutex>
#include <iostream>
#include <vector>
#include "apr_network_io.h"
#include "apr_pools.h"
#ifdef WIN32
#include "winsock.h"
#else
#include "arpa/inet.h"
#endif
#include "base/logging.h"
#include "base/network_util.h"
#include "base/network/network_util.h"
#include "command_handler/command_impl.h"
#include "device_manager.h"
#include "livox_def.h"
@@ -45,48 +38,43 @@ using std::string;
using std::vector;
using std::chrono::steady_clock;
namespace livox {
uint16_t DeviceDiscovery::port_count = 0;
bool DeviceDiscovery::Init() {
apr_status_t rv = apr_pool_create(&mem_pool_, NULL);
if (rv != APR_SUCCESS) {
LOG_ERROR(PrintAPRStatus(rv));
return false;
}
if (comm_port_ == NULL) {
comm_port_.reset(new CommPort());
}
return true;
}
bool DeviceDiscovery::Start(IOLoop *loop) {
if (loop == NULL) {
bool DeviceDiscovery::Start(std::weak_ptr<IOLoop> loop) {
if (loop.expired()) {
return false;
}
loop_ = loop;
sock_ = util::CreateBindSocket(kListenPort, mem_pool_, true);
if (sock_ == NULL) {
LOG_ERROR("DeviceDiscovery Create Socket Failed");
sock_ = util::CreateSocket(kListenPort);
if (sock_ < 0) {
return false;
}
loop_->AddDelegate(sock_, this);
loop_.lock()->AddDelegate(sock_, this);
return true;
}
void DeviceDiscovery::OnData(apr_socket_t *sock, void *) {
apr_sockaddr_t addr;
void DeviceDiscovery::OnData(socket_t sock, void *) {
struct sockaddr addr;
int addrlen = sizeof(addr);
uint32_t buf_size = 0;
uint8_t *cache_buf = comm_port_->FetchCacheFreeSpace(&buf_size);
apr_size_t size = buf_size;
apr_status_t rv = apr_socket_recvfrom(&addr, sock, 0, reinterpret_cast<char *>(cache_buf), &size);
comm_port_->UpdateCacheWrIdx(size);
if (rv != APR_SUCCESS) {
LOG_WARN(" Receive Failed {}", PrintAPRStatus(rv));
int size = buf_size;
size = util::RecvFrom(sock, reinterpret_cast<char *>(cache_buf), buf_size, 0, &addr, &addrlen);
if (size < 0) {
return;
}
comm_port_->UpdateCacheWrIdx(size);
CommPacket packet;
memset(&packet, 0, sizeof(packet));
@@ -97,10 +85,11 @@ void DeviceDiscovery::OnData(apr_socket_t *sock, void *) {
if (connecting_devices_.find(sock) == connecting_devices_.end()) {
continue;
}
DeviceInfo info = std::get<2>(connecting_devices_[sock]);
loop_->RemoveDelegate(sock, this);
apr_socket_close(sock);
apr_pool_destroy(std::get<0>(connecting_devices_[sock]));
DeviceInfo info = std::get<1>(connecting_devices_[sock]);
if (!loop_.expired()) {
loop_.lock()->RemoveDelegate(sock, this);
}
util::CloseSock(sock);
connecting_devices_.erase(sock);
if (packet.data == NULL) {
@@ -123,11 +112,12 @@ void DeviceDiscovery::OnData(apr_socket_t *sock, void *) {
void DeviceDiscovery::OnTimer(TimePoint now) {
ConnectingDeviceMap::iterator ite = connecting_devices_.begin();
while (ite != connecting_devices_.end()) {
tuple<apr_pool_t *, TimePoint, DeviceInfo> &device_tuple = ite->second;
if (now - std::get<1>(device_tuple) > std::chrono::milliseconds(500)) {
loop_->RemoveDelegate(ite->first, this);
apr_socket_close(ite->first);
apr_pool_destroy(std::get<0>(device_tuple));
tuple<TimePoint, DeviceInfo> &device_tuple = ite->second;
if (now - std::get<0>(device_tuple) > std::chrono::milliseconds(500)) {
if (!loop_.expired()) {
loop_.lock()->RemoveDelegate(ite->first, this);
}
util::CloseSock(ite->first);
connecting_devices_.erase(ite++);
} else {
++ite;
@@ -136,23 +126,20 @@ void DeviceDiscovery::OnTimer(TimePoint now) {
}
void DeviceDiscovery::Uninit() {
if (sock_) {
loop_->RemoveDelegate(sock_, this);
apr_socket_close(sock_);
sock_ = NULL;
if (sock_ > 0) {
if (!loop_.expired()) {
loop_.lock()->RemoveDelegate(sock_, this);
}
util::CloseSock(sock_);
sock_ = -1;
}
if (comm_port_) {
comm_port_.reset(NULL);
}
if (mem_pool_) {
apr_pool_destroy(mem_pool_);
mem_pool_ = NULL;
}
}
void DeviceDiscovery::OnBroadcast(const CommPacket &packet, apr_sockaddr_t *addr) {
void DeviceDiscovery::OnBroadcast(const CommPacket &packet, struct sockaddr *addr) {
if (packet.data == NULL) {
return;
}
@@ -164,11 +151,7 @@ void DeviceDiscovery::OnBroadcast(const CommPacket &packet, apr_sockaddr_t *addr
char ip[16];
memset(&ip, 0, sizeof(ip));
apr_status_t rv = apr_sockaddr_ip_getbuf(ip, sizeof(ip), addr);
if (rv != APR_SUCCESS) {
LOG_ERROR(PrintAPRStatus(rv));
return;
}
inet_ntop(AF_INET, &((struct sockaddr_in*)addr)->sin_addr, ip, INET_ADDRSTRLEN);
strncpy(device_info.ip, ip, sizeof(device_info.ip));
device_manager().BroadcastDevices(&device_info);
@@ -195,35 +178,28 @@ void DeviceDiscovery::OnBroadcast(const CommPacket &packet, apr_sockaddr_t *addr
lidar_info.status.progress = 0;
strncpy(lidar_info.ip, ip, sizeof(lidar_info.ip));
apr_pool_t *pool = NULL;
rv = apr_pool_create(&pool, mem_pool_);
if (rv != APR_SUCCESS) {
LOG_ERROR(PrintAPRStatus(rv));
socket_t cmd_sock = util::CreateSocket(lidar_info.cmd_port);
if (cmd_sock < 0) {
return;
}
apr_socket_t *cmd_sock = util::CreateBindSocket(lidar_info.cmd_port, pool);
if (cmd_sock == NULL) {
apr_pool_destroy(pool);
pool = NULL;
return;
if (!loop_.expired()) {
loop_.lock()->AddDelegate(cmd_sock, this);
}
loop_->AddDelegate(cmd_sock, this);
OnTimer(steady_clock::now());
std::get<0>(connecting_devices_[cmd_sock]) = pool;
std::get<2>(connecting_devices_[cmd_sock]) = lidar_info;
std::get<1>(connecting_devices_[cmd_sock]) = lidar_info;
bool result = false;
do {
HandshakeRequest handshake_req;
uint32_t local_ip = 0;
if (util::FindLocalIp(addr->sa.sin, local_ip) == false) {
if (util::FindLocalIp(*(struct sockaddr_in*)addr, local_ip) == false) {
result = false;
LOG_INFO("LocalIp and DeviceIp are not in same subnet");
break;
}
LOG_INFO("LocalIP: {}", inet_ntoa(*(struct in_addr *)&local_ip));
LOG_INFO("DeviceIP: {}", inet_ntoa(((struct sockaddr_in *)addr)->sin_addr));
LOG_INFO("Command Port: {}", lidar_info.cmd_port);
LOG_INFO("Data Port: {}", lidar_info.data_port);
CommPacket packet;
@@ -240,21 +216,21 @@ void DeviceDiscovery::OnBroadcast(const CommPacket &packet, apr_sockaddr_t *addr
packet.data = (uint8_t *)&handshake_req;
vector<uint8_t> buf(kMaxCommandBufferSize + 1);
apr_size_t o_len = kMaxCommandBufferSize;
int o_len = kMaxCommandBufferSize;
comm_port_->Pack(buf.data(), kMaxCommandBufferSize, (uint32_t *)&o_len, packet);
rv = apr_socket_sendto(cmd_sock, addr, 0, reinterpret_cast<const char *>(buf.data()), &o_len);
if (rv != APR_SUCCESS) {
result = false;
break;
int byte_send = sendto(cmd_sock, reinterpret_cast<const char *>(buf.data()), o_len, 0, addr, sizeof(*addr));
if (byte_send < 0) {
return;
}
std::get<1>(connecting_devices_[cmd_sock]) = steady_clock::now();
std::get<0>(connecting_devices_[cmd_sock]) = steady_clock::now();
result = true;
} while (0);
if (result == false) {
loop_->RemoveDelegate(cmd_sock, this);
apr_socket_close(cmd_sock);
apr_pool_destroy(pool);
if (!loop_.expired()) {
loop_.lock()->RemoveDelegate(cmd_sock, this);
}
util::CloseSock(cmd_sock);
connecting_devices_.erase(cmd_sock);
}
}
sdk_core/src/device_discovery.h
+22 -15
View File
@@ -27,13 +27,20 @@
#include <mutex>
#include <string>
#include "apr_general.h"
#include "apr_network_io.h"
#include "apr_pools.h"
#include <algorithm>
#include "base/io_thread.h"
#include "base/noncopyable.h"
#include "comm/comm_port.h"
#include "command_handler/command_channel.h"
#include <stdio.h>
#ifdef WIN32
#include <winsock2.h>
#include <ws2tcpip.h>
#include <ws2def.h>
#pragma comment(lib,"ws2_32.lib")
#else
#include "arpa/inet.h"
#endif
namespace livox {
@@ -45,7 +52,7 @@ class DeviceDiscovery : public noncopyable, IOLoop::IOLoopDelegate {
public:
typedef std::chrono::steady_clock::time_point TimePoint;
DeviceDiscovery() : sock_(NULL), mem_pool_(NULL), loop_(NULL), comm_port_(nullptr) {}
DeviceDiscovery() : comm_port_(nullptr) {}
bool Init();
void Uninit();
@@ -54,36 +61,36 @@ class DeviceDiscovery : public noncopyable, IOLoop::IOLoopDelegate {
* @param loop IOLoop on where DeviceDiscovery runs.
* @return true if successfully.
*/
bool Start(IOLoop *loop);
bool Start(std::weak_ptr<IOLoop> loop);
/**
* IOLoop callback delegate.
* @param client_data client data passed in IOLoop::AddDelegate
*/
void OnData(apr_socket_t *, void *client_data);
void OnData(socket_t, void *client_data);
void OnTimer(TimePoint now);
private:
void OnBroadcast(const CommPacket &packet, apr_sockaddr_t *addr);
void OnBroadcast(const CommPacket &packet, struct sockaddr *addr);
private:
/** broadcast listening port number. */
static const apr_port_t kListenPort = 55000;
static const uint16_t kListenPort = 55000;
/** command port number start offset. */
static const apr_port_t kCmdPortOffset = 500;
static const uint16_t kCmdPortOffset = 500;
/** data port number start offset. */
static const apr_port_t kDataPortOffset = 1000;
static const uint16_t kDataPortOffset = 1000;
/** sensor port number start offset. */
static const apr_port_t kSensorPortOffset = 1000;
static const uint16_t kSensorPortOffset = 1000;
static uint16_t port_count;
apr_socket_t *sock_;
apr_pool_t *mem_pool_;
IOLoop *loop_;
socket_t sock_ = -1;
std::weak_ptr<IOLoop> loop_;
std::unique_ptr<CommPort> comm_port_;
std::mutex mutex_;
typedef std::map<apr_socket_t *, std::tuple<apr_pool_t *, TimePoint, DeviceInfo> > ConnectingDeviceMap;
typedef std::map<socket_t, std::tuple<TimePoint, DeviceInfo> > ConnectingDeviceMap;
ConnectingDeviceMap connecting_devices_;
};
sdk_core/src/device_manager.cpp
-10
View File
@@ -49,11 +49,6 @@ inline bool IsLidar(uint8_t mode) {
}
bool DeviceManager::Init() {
apr_status_t rv = apr_pool_create(&mem_pool_, NULL);
if (rv != APR_SUCCESS) {
return false;
}
return true;
}
@@ -66,11 +61,6 @@ void DeviceManager::Uninit() {
for (DeviceContainer::iterator ite = devices_.begin(); ite != devices_.end(); ++ite) {
ite->clear();
}
if (mem_pool_) {
apr_pool_destroy(mem_pool_);
mem_pool_ = NULL;
}
}
bool DeviceManager::AddDevice(const DeviceInfo &device) {
sdk_core/src/device_manager.h
+2 -2
View File
@@ -29,6 +29,7 @@
#include <functional>
#include <mutex>
#include <string>
#include <string.h>
#include "device_discovery.h"
#include "livox_sdk.h"
@@ -53,7 +54,7 @@ typedef enum {
*/
class DeviceManager : public noncopyable {
public:
DeviceManager() : mem_pool_(NULL), device_mode_(kDeviceModeNone), connected_cb_(NULL), broadcast_cb_(NULL) {}
DeviceManager() : device_mode_(kDeviceModeNone), connected_cb_(NULL), broadcast_cb_(NULL) {}
bool Init();
void Uninit();
@@ -149,7 +150,6 @@ class DeviceManager : public noncopyable {
} DetailDeviceInfo;
typedef std::array<DetailDeviceInfo, kMaxConnectedDeviceNum> DeviceContainer;
DeviceContainer devices_;
apr_pool_t *mem_pool_;
DeviceMode device_mode_;
std::function<void(const DeviceInfo *, DeviceEvent)> connected_cb_;
std::function<void(const BroadcastDeviceInfo *info)> broadcast_cb_;
sdk_core/src/livox_sdk.cpp
+19 -16
View File
@@ -23,11 +23,14 @@
//
#include "livox_sdk.h"
#include "apr_general.h"
#include "command_handler/command_handler.h"
#include "data_handler/data_handler.h"
#include "base/logging.h"
#include "device_manager.h"
#ifdef WIN32
#include<winsock2.h>
#endif // WIN32
using namespace livox;
IOThread *g_thread = NULL;
static bool is_initialized = false;
@@ -44,16 +47,18 @@ bool Init() {
if (is_initialized) {
return false;
}
#ifdef WIN32
WORD sockVersion = MAKEWORD(2, 0);
WSADATA wsdata;
if (WSAStartup(sockVersion, &wsdata) != 0) {
return false;
}
#endif // WIN32
bool result = false;
do {
InitLogger();
if (apr_initialize() != APR_SUCCESS) {
result = false;
break;
}
g_thread = new IOThread();
g_thread->Init();
@@ -79,10 +84,6 @@ bool Init() {
result = true;
} while (0);
if (result == false) {
apr_terminate();
}
is_initialized = result;
return result;
}
@@ -91,24 +92,26 @@ void Uninit() {
if (!is_initialized) {
return;
}
#ifdef WIN32
WSACleanup();
#endif // WIN32
if (g_thread) {
g_thread->Quit();
g_thread->Join();
}
device_discovery().Uninit();
command_handler().Uninit();
data_handler().Uninit();
device_manager().Uninit();
if (g_thread) {
g_thread->Uninit();
delete g_thread;
g_thread = NULL;
}
device_discovery().Uninit();
command_handler().Uninit();
data_handler().Uninit();
device_manager().Uninit();
UninitLogger();
apr_terminate();
is_initialized = false;
}