Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions Common/Cpp/StreamConnections/PushingStreamConnections.h
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,11 @@ class ReliableStreamConnectionPushing : public StreamConnectionPushing{
WallClock deadline
) = 0;

// Wait until the remote has acked everything sent so far.
// Returns false if this doesn't happen within the timeout.
// Throws exception if connection is dead.
virtual bool wait_for_pending(WallDuration timeout) = 0;

protected:
void on_reliable_recv(const void* data, size_t bytes){
m_listeners.run_method(&StreamListener::on_recv, data, bytes);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ class ReliableStreamConnection final
}

size_t pending() const;
bool wait_for_pending(WallDuration timeout = WallDuration::max());
virtual bool wait_for_pending(WallDuration timeout = WallDuration::max()) override;


public:
Expand Down
12 changes: 12 additions & 0 deletions SerialPrograms/Source/Controllers/Controller.h
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,18 @@ class AbstractController{
// ever releasing it during the transition.
virtual void replace_on_next_command() = 0;

// Same as "cancel_all_commands()", but unlike the two functions above,
// this one blocks until the device acks the cancel or until "timeout".
//
// Returns true if the device has received the cancel. Commands issued
// after this returns are queued behind the cancel, so the controller is
// immediately usable.
// Returns false on timeout or if the controller cannot confirm cancels.
//
// Commands issued from other threads while this is running may land
// after the cancel and are not cancelled.
virtual bool cancel_all_and_wait(Milliseconds timeout) = 0;


public:
//
Expand Down
3 changes: 3 additions & 0 deletions SerialPrograms/Source/Controllers/NullController.h
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,9 @@ class NullController final : public AbstractController{

virtual void cancel_all_commands() override{}
virtual void replace_on_next_command() override{}
virtual bool cancel_all_and_wait(Milliseconds timeout) override{
return false;
}
virtual void wait_for_all(Cancellable* cancellable) override{}
virtual void issue_barrier(Cancellable* cancellable) override{}
virtual void issue_nop(Cancellable* cancellable, Milliseconds duration) override{}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@ void CommandQueueManager::set_command_queue_size(uint8_t command_queue_size){
}
m_cv.notify_all();
}


bool CommandQueueManager::cancel(std::exception_ptr exception) noexcept{
bool ret = Cancellable::cancel(std::move(exception));
{
Expand All @@ -39,13 +41,16 @@ bool CommandQueueManager::cancel(std::exception_ptr exception) noexcept{
return ret;
}


void CommandQueueManager::wait_for_all(Cancellable* cancellable){
std::unique_lock<Mutex> lg(m_lock);
while (!m_pending_commands.empty()){
throw_if_cancelled(cancellable);
cv_wait(cancellable, lg);
}
}


void CommandQueueManager::wait_for_command_finish(Cancellable* cancellable, uint8_t id){
std::unique_lock<Mutex> lg(m_lock);
while (true){
Expand Down Expand Up @@ -81,6 +86,41 @@ CommandQueueManager::CancelResult CommandQueueManager::send_cancel() noexcept{
m_logger.log("cancel_all_commands() - Buffer Full", COLOR_RED);
return CancelResult::BUFFER_FULL;
}


bool CommandQueueManager::wait_for_cancel(WallDuration timeout){
WallClock deadline = current_time() + timeout;
{
std::unique_lock<Mutex> lg(m_lock);

// If "send_cancel()" hit a full buffer, the cancel hasn't been sent
// yet. Keep retrying as the device acks what's in flight.
while (m_pending_special == PABB2_MESSAGE_OPCODE_CQ_CANCEL &&
!try_push_pending_specials()
){
WallClock now = current_time();
if (now >= deadline){
m_logger.log("wait_for_cancel(): Timed out sending the cancel.", COLOR_RED);
return false;
}
lg.unlock();
m_connection.wait_for_pending(deadline - now);
lg.lock();
}
}
m_cv.notify_all();

// The device acks a packet once it lands in its receive buffer. Commands
// are processed in order, so anything sent after this is behind the cancel.
WallClock now = current_time();
if (m_connection.wait_for_pending(now < deadline ? deadline - now : WallDuration::zero())){
return true;
}
m_logger.log("wait_for_cancel(): Timed out waiting for the device to ack the cancel.", COLOR_RED);
return false;
}


CommandQueueManager::CancelResult CommandQueueManager::send_replace_on_next() noexcept{
bool success;
{
Expand Down Expand Up @@ -157,6 +197,8 @@ uint8_t CommandQueueManager::send_command(Cancellable* cancellable, MessageHeade
m_cv.notify_all();
return command.id;
}


void CommandQueueManager::report_command_finished(const MessageHeader& finished_message){
{
std::lock_guard<Mutex> lg(m_lock);
Expand Down Expand Up @@ -224,6 +266,8 @@ void CommandQueueManager::on_cancellable_cancel(
}
m_cv.notify_all();
}


void CommandQueueManager::cv_wait(Cancellable* cancellable, std::unique_lock<Mutex>& lg){
if (cancellable == nullptr){
m_cv.wait(lg);
Expand All @@ -239,6 +283,8 @@ void CommandQueueManager::cv_wait(Cancellable* cancellable, std::unique_lock<Mut
cancellable->remove_cancel_listener(*this);
lg.lock();
}


void CommandQueueManager::throw_if_cancelled(Cancellable* cancellable){
Cancellable::throw_if_cancelled();
if (cancellable){
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,11 @@ class CommandQueueManager final : public Cancellable, public Cancellable::Cancel
CancelResult send_cancel() noexcept;
CancelResult send_replace_on_next() noexcept;

// Call after "send_cancel()". Blocks until the device has acked the
// cancel (and everything sent before it) or until the timeout.
// Returns true if acked. Throws if the connection is dead.
bool wait_for_cancel(WallDuration timeout);

uint8_t send_command(Cancellable* cancellable, MessageHeader& command);
void report_command_finished(const MessageHeader& finished_message);

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,12 @@ void PABotBase2_Keyboard::replace_on_next_command(){
m_connection.device().command_queue().send_replace_on_next();
m_scheduler.clear_on_next();
}
bool PABotBase2_Keyboard::cancel_all_and_wait(Milliseconds timeout){
cancel_all_commands();

// Wait outside "m_state_lock" since this can take a while.
return m_connection.device().command_queue().wait_for_cancel(timeout);
}


void PABotBase2_Keyboard::wait_for_all(Cancellable* cancellable){
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,7 @@ class PABotBase2_Keyboard final :
public:
virtual void cancel_all_commands() override;
virtual void replace_on_next_command() override;
virtual bool cancel_all_and_wait(Milliseconds timeout) override;

virtual void wait_for_all(Cancellable* cancellable) override;

Expand Down
9 changes: 9 additions & 0 deletions SerialPrograms/Source/Integrations/PybindSwitchController.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -145,6 +145,15 @@ void PybindSwitchProController::wait_for_all_requests(){
}
controller->wait_for_all(nullptr);
}
bool PybindSwitchProController::cancel_all_and_wait(uint64_t timeout_millis){
PybindSwitchProControllerInternal* internal = (PybindSwitchProControllerInternal*)m_internals;
ProController* controller = internal->controller_if_ready();
if (controller == nullptr){
internal->m_logger.log("Controller is not ready.", COLOR_RED);
return false;
}
return controller->cancel_all_and_wait(Milliseconds(timeout_millis));
}
void PybindSwitchProController::wait(uint64_t duration){
PybindSwitchProControllerInternal* internal = (PybindSwitchProControllerInternal*)m_internals;
internal->controller().issue_nop(nullptr, Milliseconds(duration));
Expand Down
13 changes: 12 additions & 1 deletion SerialPrograms/Source/Integrations/PybindSwitchController.h
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,9 @@ namespace NintendoSwitch{
// `NintendoSwitch::DpadPosition` values (0 = up, clockwise to 7 = up-left, 8 = none).
// Joystick coordinates are in [-1.0, 1.0] with +x = right and +y = up.
//
// Thread safety: all methods can be called from any thread.
// Thread safety: all methods can be called from any thread. In particular
// `cancel_all_and_wait()` may be called while another thread is blocked in
// `wait_for_all_requests()`, e.g. to stop everything in an emergency.
class PybindSwitchProController{
PybindSwitchProController(const PybindSwitchProController&) = delete;
void operator=(const PybindSwitchProController&) = delete;
Expand Down Expand Up @@ -68,6 +70,15 @@ class PybindSwitchProController{
// Returns immediately if the controller is not ready.
void wait_for_all_requests();

// Drop every queued command that has not executed yet, return to the neutral
// state (no buttons pressed, sticks centered), and wait at most `timeout_millis`
// for the device to acknowledge that it received the cancel.
//
// Returns true if acknowledged, false on timeout or if the controller is not ready.
// Commands pushed from other threads while this is running may land after the
// cancel and are not cancelled.
bool cancel_all_and_wait(uint64_t timeout_millis);

public:
// Commands. These throw InvalidConnectionStateException if the controller is not
// ready, and block only if the device's command queue is full.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,12 @@ void PABotBase2_Controller::replace_on_next_command(){
m_connection.device().command_queue().send_replace_on_next();
m_scheduler.clear_on_next();
}
bool PABotBase2_Controller::cancel_all_and_wait(Milliseconds timeout){
cancel_all_commands();

// Wait outside "m_state_lock" since this can take a while.
return m_connection.device().command_queue().wait_for_cancel(timeout);
}


void PABotBase2_Controller::wait_for_all(Cancellable* cancellable){
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ class PABotBase2_Controller : public ControllerWithScheduler{
public:
void cancel_all_commands();
void replace_on_next_command();
bool cancel_all_and_wait(Milliseconds timeout);

void wait_for_all(Cancellable* cancellable);

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,9 @@ class PABotBase2_JoyCon :
virtual void replace_on_next_command() override{
PABotBase2_Controller::replace_on_next_command();
}
virtual bool cancel_all_and_wait(Milliseconds timeout) override{
return PABotBase2_Controller::cancel_all_and_wait(timeout);
}

virtual void wait_for_all(Cancellable* cancellable) override{
PABotBase2_Controller::wait_for_all(cancellable);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,9 @@ class PABotBase2_ProController final :
virtual void replace_on_next_command() override{
PABotBase2_Controller::replace_on_next_command();
}
virtual bool cancel_all_and_wait(Milliseconds timeout) override{
return PABotBase2_Controller::cancel_all_and_wait(timeout);
}

virtual void wait_for_all(Cancellable* cancellable) override{
PABotBase2_Controller::wait_for_all(cancellable);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,9 @@ class PABotBase2_WiredController final :
virtual void replace_on_next_command() override{
PABotBase2_Controller::replace_on_next_command();
}
virtual bool cancel_all_and_wait(Milliseconds timeout) override{
return PABotBase2_Controller::cancel_all_and_wait(timeout);
}

virtual void wait_for_all(Cancellable* cancellable) override{
PABotBase2_Controller::wait_for_all(cancellable);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,11 @@ class ProController_SysbotBase3 final :
public:
virtual void cancel_all_commands() override;
virtual void replace_on_next_command() override;
virtual bool cancel_all_and_wait(Milliseconds timeout) override{
// sys-botbase does not ack cancels.
cancel_all_commands();
return false;
}

virtual void wait_for_all(Cancellable* cancellable) override;

Expand Down
Loading