Refactoring and small improvements to async rpc operations.

Added AsyncRPCQueue::closeAndWait() so rpcserver can block on worker threads when shutting down.
AsyncRPCOperation is no longer copyable - copy constructor and assignment operators now private.
Refactoring: renamed methods, renamed member variables
Tidy up: comments, const, size_t, braces
This commit is contained in:
Simon 2016-08-28 19:50:39 -07:00
parent 34f0001ccc
commit 3b54bf5813
6 changed files with 272 additions and 207 deletions

View File

@ -25,25 +25,27 @@ std::map<OperationStatus, std::string> OperationStatusMap = {
{OperationStatus::SUCCESS, "success"} {OperationStatus::SUCCESS, "success"}
}; };
AsyncRPCOperation::AsyncRPCOperation() : errorCode(0), errorMessage() { /**
* Every operation instance should have a globally unique id
*/
AsyncRPCOperation::AsyncRPCOperation() : error_code_(0), error_message_() {
// Set a unique reference for each operation // Set a unique reference for each operation
boost::uuids::uuid uuid = uuidgen(); boost::uuids::uuid uuid = uuidgen();
std::string s = "opid-" + boost::uuids::to_string(uuid); std::string s = "opid-" + boost::uuids::to_string(uuid);
setId(s); set_id(s);
setState(OperationStatus::READY); set_state(OperationStatus::READY);
creationTime = (int64_t)time(NULL); creation_time_ = (int64_t)time(NULL);
} }
AsyncRPCOperation::AsyncRPCOperation(const AsyncRPCOperation& o) : id(o.id), creationTime(o.creationTime), state(o.state.load()) AsyncRPCOperation::AsyncRPCOperation(const AsyncRPCOperation& o) : id_(o.id_), creation_time_(o.creation_time_), state_(o.state_.load())
{ {
} }
AsyncRPCOperation& AsyncRPCOperation::operator=( const AsyncRPCOperation& other ) { AsyncRPCOperation& AsyncRPCOperation::operator=( const AsyncRPCOperation& other ) {
this->id = other.getId(); this->id_ = other.getId();
this->creationTime = other.creationTime; this->creation_time_ = other.creation_time_;
this->state.store(other.state.load()); this->state_.store(other.state_.load());
return *this; return *this;
} }
@ -51,73 +53,85 @@ AsyncRPCOperation& AsyncRPCOperation::operator=( const AsyncRPCOperation& other
AsyncRPCOperation::~AsyncRPCOperation() { AsyncRPCOperation::~AsyncRPCOperation() {
} }
/**
* Override this cancel() method if you can interrupt main() when executing.
*/
void AsyncRPCOperation::cancel() { void AsyncRPCOperation::cancel() {
if (isReady()) if (isReady()) {
setState(OperationStatus::CANCELLED); set_state(OperationStatus::CANCELLED);
}
} }
/**
void AsyncRPCOperation::startExecutionClock() { * Start timing the execution run of the code you're interested in
startTime = std::chrono::system_clock::now(); */
void AsyncRPCOperation::start_execution_clock() {
start_time_ = std::chrono::system_clock::now();
} }
void AsyncRPCOperation::stopExecutionClock() { /**
endTime = std::chrono::system_clock::now(); * Stop timing the execution run
*/
void AsyncRPCOperation::stop_execution_clock() {
end_time_ = std::chrono::system_clock::now();
} }
/**
// Implement this method in any subclass. * Implement this virtual method in any subclass. This is just an example implementation.
// This is just an example implementation. */
void AsyncRPCOperation::main() { void AsyncRPCOperation::main() {
if (isCancelled()) if (isCancelled()) {
return; return;
setState(OperationStatus::EXECUTING);
//
// Do some work here...
//
startExecutionClock();
//std::this_thread::sleep_for(std::chrono::milliseconds(10000));
stopExecutionClock();
// If there was an error...
// setErrorCode(123);
// setErrorMessage("Murphy's law");
// setState(OperationStatus::FAILED);
// Otherwise
Value v("We have a result!");
setResult(v);
setState(OperationStatus::SUCCESS);
} }
set_state(OperationStatus::EXECUTING);
start_execution_clock();
// Do some work here..
stop_execution_clock();
// If there was an error, you might set it like this:
/*
setErrorCode(123);
setErrorMessage("Murphy's law");
setState(OperationStatus::FAILED);
*/
// Otherwise, if the operation was a success:
Value v("We have a result!");
set_result(v);
set_state(OperationStatus::SUCCESS);
}
/**
* Return the error of the completed operation as a Value object.
*/
Value AsyncRPCOperation::getError() const { Value AsyncRPCOperation::getError() const {
if (!isFailed()) if (!isFailed()) {
return Value::null; return Value::null;
}
Object error; Object error;
error.push_back(Pair("code", this->errorCode)); error.push_back(Pair("code", this->error_code_));
error.push_back(Pair("message", this->errorMessage)); error.push_back(Pair("message", this->error_message_));
return Value(error); return Value(error);
} }
/**
* Return the result of the completed operation as a Value object.
*/
Value AsyncRPCOperation::getResult() const { Value AsyncRPCOperation::getResult() const {
if (!isSuccess()) if (!isSuccess()) {
return Value::null; return Value::null;
}
return this->resultValue; return this->result_;
} }
/* /**
* Returns a status Value object. * Returns a status Value object.
* If the operation has failed, it will include an error object. * If the operation has failed, it will include an error object.
* If the operation has succeeded, it will include the result value. * If the operation has succeeded, it will include the result value.
@ -127,7 +141,7 @@ Value AsyncRPCOperation::getStatus() const {
Object obj; Object obj;
obj.push_back(Pair("id", this->getId())); obj.push_back(Pair("id", this->getId()));
obj.push_back(Pair("status", OperationStatusMap[status])); obj.push_back(Pair("status", OperationStatusMap[status]));
obj.push_back(Pair("creation_time", this->creationTime)); obj.push_back(Pair("creation_time", this->creation_time_));
// creation, exec time, duration, exec end, etc. // creation, exec time, duration, exec end, etc.
Value err = this->getError(); Value err = this->getError();
if (!err.is_null()) { if (!err.is_null()) {
@ -138,14 +152,16 @@ Value AsyncRPCOperation::getStatus() const {
obj.push_back(Pair("result", result)); obj.push_back(Pair("result", result));
// Include execution time for successful operation // Include execution time for successful operation
std::chrono::duration<double> elapsed_seconds = endTime - startTime; std::chrono::duration<double> elapsed_seconds = end_time_ - start_time_;
obj.push_back(Pair("execution_secs", elapsed_seconds.count())); obj.push_back(Pair("execution_secs", elapsed_seconds.count()));
} }
return Value(obj); return Value(obj);
} }
/**
* Return the operation state in human readable form.
*/
std::string AsyncRPCOperation::getStateAsString() const { std::string AsyncRPCOperation::getStateAsString() const {
OperationStatus status = this->getState(); OperationStatus status = this->getState();
return OperationStatusMap[status]; return OperationStatusMap[status];

View File

@ -20,11 +20,11 @@ using namespace std;
using namespace json_spirit; using namespace json_spirit;
/** /**
* AsyncRPCOperations are given to the AsyncRPCQueue for processing. * AsyncRPCOperation objects are submitted to the AsyncRPCQueue for processing.
* *
* How to subclass: * To subclass AsyncRPCOperation, implement the main() method.
* Implement the main() method, this is where work is performed.
* Update the operation status as work is underway and completes. * Update the operation status as work is underway and completes.
* If main() can be interrupted, inmplement the cancel() method.
*/ */
typedef std::string AsyncRPCOperationId; typedef std::string AsyncRPCOperationId;
@ -40,30 +40,26 @@ typedef enum class operationStateEnum {
class AsyncRPCOperation { class AsyncRPCOperation {
public: public:
AsyncRPCOperation(); AsyncRPCOperation();
// Todo: keep or delete copy constructors and assignment?
AsyncRPCOperation(const AsyncRPCOperation& orig);
AsyncRPCOperation& operator=( const AsyncRPCOperation& other );
virtual ~AsyncRPCOperation(); virtual ~AsyncRPCOperation();
// Implement this method in your subclass. // You must implement this method in your subclass.
virtual void main(); virtual void main();
// Override this method if you can interrupt execution of main() in your subclass.
void cancel(); void cancel();
// Getters and setters // Getters and setters
OperationStatus getState() const { OperationStatus getState() const {
return state.load(); return state_.load();
} }
AsyncRPCOperationId getId() const { AsyncRPCOperationId getId() const {
return id; return id_;
} }
int64_t getCreationTime() const { int64_t getCreationTime() const {
return creationTime; return creation_time_;
} }
Value getStatus() const; Value getStatus() const;
@ -75,11 +71,11 @@ public:
std::string getStateAsString() const; std::string getStateAsString() const;
int getErrorCode() const { int getErrorCode() const {
return errorCode; return error_code_;
} }
std::string getErrorMessage() const { std::string getErrorMessage() const {
return errorMessage; return error_message_;
} }
bool isCancelled() const { bool isCancelled() const {
@ -104,46 +100,47 @@ public:
protected: protected:
Value resultValue; Value result_;
int errorCode; int error_code_;
std::string errorMessage; std::string error_message_;
std::atomic<OperationStatus> state; std::atomic<OperationStatus> state_;
std::chrono::time_point<std::chrono::system_clock> startTime, endTime; std::chrono::time_point<std::chrono::system_clock> start_time_, end_time_;
void startExecutionClock(); void start_execution_clock();
void stopExecutionClock(); void stop_execution_clock();
void setState(OperationStatus state) { void set_state(OperationStatus state) {
this->state.store(state); this->state_.store(state);
} }
void setErrorCode(int errorCode) { void set_error_code(int errorCode) {
this->errorCode = errorCode; this->error_code_ = errorCode;
} }
void setErrorMessage(std::string errorMessage) { void set_error_message(std::string errorMessage) {
this->errorMessage = errorMessage; this->error_message_ = errorMessage;
} }
void setResult(Value v) { void set_result(Value v) {
this->resultValue = v; this->result_ = v;
} }
private: private:
// Todo: Private for now. If copying an operation is possible, should it // Derived classes should write their own copy constructor and assignment operators
// receive a new id and a new creation time? AsyncRPCOperation(const AsyncRPCOperation& orig);
void setId(AsyncRPCOperationId id) { AsyncRPCOperation& operator=( const AsyncRPCOperation& other );
this->id = id;
void set_id(AsyncRPCOperationId id) {
this->id_ = id;
} }
// Todo: Ditto above. void set_creation_time(int64_t creationTime) {
void setCreationTime(int64_t creationTime) { this->creation_time_ = creationTime;
this->creationTime = creationTime;
} }
AsyncRPCOperationId id; AsyncRPCOperationId id_;
int64_t creationTime; int64_t creation_time_;
}; };
#endif /* ASYNCRPCOPERATION_H */ #endif /* ASYNCRPCOPERATION_H */

View File

@ -4,47 +4,41 @@
#include "asyncrpcqueue.h" #include "asyncrpcqueue.h"
static std::atomic<int> workerCounter(0); static std::atomic<size_t> workerCounter(0);
AsyncRPCQueue::AsyncRPCQueue() : closed(false) { AsyncRPCQueue::AsyncRPCQueue() : closed_(false) {
} }
/*
* Calling thread will join on all the worker threads
*/
AsyncRPCQueue::~AsyncRPCQueue() { AsyncRPCQueue::~AsyncRPCQueue() {
this->closed = true; // set this in case close() was not invoked closeAndWait(); // join on all worker threads
for (std::thread & t : this->workers) {
t.join();
}
} }
/* /**
* A worker will execute this method on a new thread * A worker will execute this method on a new thread
*/ */
void AsyncRPCQueue::run(int workerId) { void AsyncRPCQueue::run(size_t workerId) {
// std::cout << "Launched queue worker " << workerId << std::endl;
while (!isClosed()) { while (!isClosed()) {
AsyncRPCOperationId key; AsyncRPCOperationId key;
std::shared_ptr<AsyncRPCOperation> operation; std::shared_ptr<AsyncRPCOperation> operation;
{ {
std::unique_lock< std::mutex > guard(cs_lock); std::unique_lock< std::mutex > guard(lock_);
while (operationIdQueue.empty() && !isClosed()) { while (operation_id_queue_.empty() && !isClosed()) {
this->cs_condition.wait(guard); this->condition_.wait(guard);
} }
// Exit if the queue is closing. // Exit if the queue is closing.
if (isClosed()) if (isClosed()) {
break; break;
}
// Get operation id // Get operation id
key = operationIdQueue.front(); key = operation_id_queue_.front();
operationIdQueue.pop(); operation_id_queue_.pop();
// Search operation map // Search operation map
AsyncRPCOperationMap::const_iterator iter = operationMap.find(key); AsyncRPCOperationMap::const_iterator iter = operation_map_.find(key);
if (iter != operationMap.end()) { if (iter != operation_map_.end()) {
operation = iter->second; operation = iter->second;
} }
} }
@ -57,15 +51,14 @@ void AsyncRPCQueue::run(int workerId) {
operation->main(); operation->main();
} }
} }
// std::cout << "Terminating queue worker " << workerId << std::endl;
} }
/* /**
* Add shared_ptr to operation. * Add shared_ptr to operation.
* *
* To retain polymorphic behaviour, i.e. main() method of derived classes is invoked, * To retain polymorphic behaviour, i.e. main() method of derived classes is invoked,
* caller should create the shared_ptr like thi: * caller should create the shared_ptr like this:
* *
* std::shared_ptr<AsyncRPCOperation> ptr(new MyCustomAsyncRPCOperation(params)); * std::shared_ptr<AsyncRPCOperation> ptr(new MyCustomAsyncRPCOperation(params));
* *
@ -74,84 +67,116 @@ void AsyncRPCQueue::run(int workerId) {
void AsyncRPCQueue::addOperation(const std::shared_ptr<AsyncRPCOperation> &ptrOperation) { void AsyncRPCQueue::addOperation(const std::shared_ptr<AsyncRPCOperation> &ptrOperation) {
// Don't add if queue is closed // Don't add if queue is closed
if (isClosed()) if (isClosed()) {
return; return;
}
AsyncRPCOperationId id = ptrOperation->getId(); AsyncRPCOperationId id = ptrOperation->getId();
{ std::lock_guard< std::mutex > guard(lock_);
std::lock_guard< std::mutex > guard(cs_lock); operation_map_.emplace(id, ptrOperation);
operationMap.emplace(id, ptrOperation); operation_id_queue_.push(id);
operationIdQueue.push(id); this->condition_.notify_one();
this->cs_condition.notify_one();
}
} }
/**
std::shared_ptr<AsyncRPCOperation> AsyncRPCQueue::getOperationForId(AsyncRPCOperationId id) { * Return the operation for a given operation id.
*/
std::shared_ptr<AsyncRPCOperation> AsyncRPCQueue::getOperationForId(AsyncRPCOperationId id) const {
std::shared_ptr<AsyncRPCOperation> ptr; std::shared_ptr<AsyncRPCOperation> ptr;
std::lock_guard< std::mutex > guard(cs_lock); std::lock_guard< std::mutex > guard(lock_);
AsyncRPCOperationMap::const_iterator iter = operationMap.find(id); AsyncRPCOperationMap::const_iterator iter = operation_map_.find(id);
if (iter != operationMap.end()) { if (iter != operation_map_.end()) {
ptr = iter->second; ptr = iter->second;
} }
return ptr; return ptr;
} }
/**
* Return the operation for a given operation id and then remove the operation from internal storage.
*/
std::shared_ptr<AsyncRPCOperation> AsyncRPCQueue::popOperationForId(AsyncRPCOperationId id) { std::shared_ptr<AsyncRPCOperation> AsyncRPCQueue::popOperationForId(AsyncRPCOperationId id) {
std::shared_ptr<AsyncRPCOperation> ptr = getOperationForId(id); std::shared_ptr<AsyncRPCOperation> ptr = getOperationForId(id);
if (ptr) { if (ptr) {
std::lock_guard< std::mutex > guard(cs_lock); std::lock_guard< std::mutex > guard(lock_);
// Note: if the id still exists in the operationIdQueue, when it gets processed by a worker // Note: if the id still exists in the operationIdQueue, when it gets processed by a worker
// there will no operation in the map to execute, so nothing will happen. // there will no operation in the map to execute, so nothing will happen.
operationMap.erase(id); operation_map_.erase(id);
} }
return ptr; return ptr;
} }
bool AsyncRPCQueue::isClosed() { /**
return closed; * Return true if the queue is closed to new operations.
*/
bool AsyncRPCQueue::isClosed() const {
return closed_;
} }
/**
* Close the queue and cancel all existing operations
*/
void AsyncRPCQueue::close() { void AsyncRPCQueue::close() {
this->closed = true; this->closed_ = true;
cancelAllOperations(); cancelAllOperations();
} }
/* /**
* Call cancel() on all operations * Call cancel() on all operations
*/ */
void AsyncRPCQueue::cancelAllOperations() { void AsyncRPCQueue::cancelAllOperations() {
std::unique_lock< std::mutex > guard(cs_lock); std::unique_lock< std::mutex > guard(lock_);
for (auto key : operationMap) { for (auto key : operation_map_) {
key.second->cancel(); key.second->cancel();
} }
this->cs_condition.notify_all(); this->condition_.notify_all();
} }
int AsyncRPCQueue::getOperationCount() { /**
std::unique_lock< std::mutex > guard(cs_lock); * Return the number of operations in the queue
return operationIdQueue.size(); */
size_t AsyncRPCQueue::getOperationCount() const {
std::unique_lock< std::mutex > guard(lock_);
return operation_id_queue_.size();
} }
/* /**
* Spawn a worker thread * Spawn a worker thread
*/ */
void AsyncRPCQueue::addWorker() { void AsyncRPCQueue::addWorker() {
std::unique_lock< std::mutex > guard(cs_lock); // Todo: could just have a lock on the vector std::unique_lock< std::mutex > guard(lock_); // Todo: could just have a lock on the vector
workers.emplace_back( std::thread(&AsyncRPCQueue::run, this, ++workerCounter) ); workers_.emplace_back( std::thread(&AsyncRPCQueue::run, this, ++workerCounter) );
} }
int AsyncRPCQueue::getNumberOfWorkers() { /**
return workers.size(); * Return the number of worker threads spawned by the queue
*/
size_t AsyncRPCQueue::getNumberOfWorkers() const {
return workers_.size();
} }
/**
std::vector<AsyncRPCOperationId> AsyncRPCQueue::getAllOperationIds() { * Return a list of all known operation ids found in internal storage.
std::unique_lock< std::mutex > guard(cs_lock); */
std::vector<AsyncRPCOperationId> AsyncRPCQueue::getAllOperationIds() const {
std::unique_lock< std::mutex > guard(lock_);
std::vector<AsyncRPCOperationId> v; std::vector<AsyncRPCOperationId> v;
for(auto & entry: operationMap) for(auto & entry: operation_map_) {
v.push_back(entry.first); v.push_back(entry.first);
}
return v; return v;
} }
/**
* Calling thread will close and wait for worker threads to join.
*/
void AsyncRPCQueue::closeAndWait() {
if (!this->closed_) {
close();
}
for (std::thread & t : this->workers_) {
if (t.joinable()) {
t.join();
}
}
}

View File

@ -16,7 +16,6 @@
#include <future> #include <future>
#include <thread> #include <thread>
#include <utility> #include <utility>
#include <memory> #include <memory>
@ -35,26 +34,28 @@ public:
AsyncRPCQueue& operator=(AsyncRPCQueue &&) = delete; // Move assign AsyncRPCQueue& operator=(AsyncRPCQueue &&) = delete; // Move assign
void addWorker(); void addWorker();
int getNumberOfWorkers(); size_t getNumberOfWorkers() const;
bool isClosed(); bool isClosed() const;
void close(); void close();
void closeAndWait();
void cancelAllOperations(); void cancelAllOperations();
int getOperationCount(); size_t getOperationCount() const;
std::shared_ptr<AsyncRPCOperation> getOperationForId(AsyncRPCOperationId); std::shared_ptr<AsyncRPCOperation> getOperationForId(AsyncRPCOperationId) const;
std::shared_ptr<AsyncRPCOperation> popOperationForId(AsyncRPCOperationId); std::shared_ptr<AsyncRPCOperation> popOperationForId(AsyncRPCOperationId);
void addOperation(const std::shared_ptr<AsyncRPCOperation> &ptrOperation); void addOperation(const std::shared_ptr<AsyncRPCOperation> &ptrOperation);
std::vector<AsyncRPCOperationId> getAllOperationIds(); std::vector<AsyncRPCOperationId> getAllOperationIds() const;
private: private:
// addWorker() will spawn a new thread on this method // addWorker() will spawn a new thread on this method
void run(int workerId); void run(size_t workerId);
std::mutex cs_lock; // Why this is not a recursive lock: http://www.zaval.org/resources/library/butenhof1.html
std::condition_variable cs_condition; mutable std::mutex lock_;
bool closed; std::condition_variable condition_;
AsyncRPCOperationMap operationMap; bool closed_;
std::queue <AsyncRPCOperationId> operationIdQueue; AsyncRPCOperationMap operation_map_;
std::vector<std::thread> workers; std::queue <AsyncRPCOperationId> operation_id_queue_;
std::vector<std::thread> workers_;
}; };
#endif #endif

View File

@ -809,8 +809,8 @@ void StopRPCThreads()
delete rpc_io_service; rpc_io_service = NULL; delete rpc_io_service; rpc_io_service = NULL;
// Tells async queue to cancel all operations and shutdown. // Tells async queue to cancel all operations and shutdown.
// The async queue destructor will block and join on worker threads. LogPrintf("%s: waiting for async rpc workers to stop\n", __func__);
async_rpc_queue->close(); async_rpc_queue->closeAndWait();
} }
bool IsRPCRunning() bool IsRPCRunning()

View File

@ -36,14 +36,17 @@ AsyncRPCOperation_sendmany::AsyncRPCOperation_sendmany(
int minDepth) : int minDepth) :
fromaddress_(fromAddress), t_outputs_(tOutputs), z_outputs_(zOutputs), mindepth_(minDepth) fromaddress_(fromAddress), t_outputs_(tOutputs), z_outputs_(zOutputs), mindepth_(minDepth)
{ {
if (minDepth < 0) if (minDepth < 0) {
throw JSONRPCError(RPC_INVALID_PARAMETER, "Minconf cannot be negative"); throw JSONRPCError(RPC_INVALID_PARAMETER, "Minconf cannot be negative");
}
if (fromAddress.size() == 0) if (fromAddress.size() == 0) {
throw JSONRPCError(RPC_INVALID_PARAMETER, "From address parameter missing"); throw JSONRPCError(RPC_INVALID_PARAMETER, "From address parameter missing");
}
if (tOutputs.size() == 0 && zOutputs.size() == 0) if (tOutputs.size() == 0 && zOutputs.size() == 0) {
throw JSONRPCError(RPC_INVALID_PARAMETER, "No recipients"); throw JSONRPCError(RPC_INVALID_PARAMETER, "No recipients");
}
fromtaddr_ = CBitcoinAddress(fromAddress); fromtaddr_ = CBitcoinAddress(fromAddress);
isfromtaddr_ = fromtaddr_.IsValid(); isfromtaddr_ = fromtaddr_.IsValid();
@ -56,12 +59,14 @@ AsyncRPCOperation_sendmany::AsyncRPCOperation_sendmany(
PaymentAddress addr = address.Get(); PaymentAddress addr = address.Get();
// We don't need to lock on the wallet as spending key related methods are thread-safe // We don't need to lock on the wallet as spending key related methods are thread-safe
if (!pwalletMain->HaveSpendingKey(addr)) if (!pwalletMain->HaveSpendingKey(addr)) {
throw JSONRPCError(RPC_INVALID_ADDRESS_OR_KEY, "Invalid from address, should be a taddr or zaddr."); throw JSONRPCError(RPC_INVALID_ADDRESS_OR_KEY, "Invalid from address, should be a taddr or zaddr.");
}
SpendingKey key; SpendingKey key;
if (!pwalletMain->GetSpendingKey(addr, key)) if (!pwalletMain->GetSpendingKey(addr, key)) {
throw JSONRPCError(RPC_INVALID_ADDRESS_OR_KEY, "Invalid from address, no spending key found for zaddr"); throw JSONRPCError(RPC_INVALID_ADDRESS_OR_KEY, "Invalid from address, no spending key found for zaddr");
}
isfromzaddr_ = true; isfromzaddr_ = true;
frompaymentaddress_ = addr; frompaymentaddress_ = addr;
@ -79,8 +84,8 @@ void AsyncRPCOperation_sendmany::main() {
if (isCancelled()) if (isCancelled())
return; return;
setState(OperationStatus::EXECUTING); set_state(OperationStatus::EXECUTING);
startExecutionClock(); start_execution_clock();
bool success = false; bool success = false;
@ -89,25 +94,25 @@ void AsyncRPCOperation_sendmany::main() {
} catch (Object objError) { } catch (Object objError) {
int code = find_value(objError, "code").get_int(); int code = find_value(objError, "code").get_int();
std::string message = find_value(objError, "message").get_str(); std::string message = find_value(objError, "message").get_str();
setErrorCode(code); set_error_code(code);
setErrorMessage(message); set_error_message(message);
} catch (runtime_error e) { } catch (runtime_error e) {
setErrorCode(-1); set_error_code(-1);
setErrorMessage("runtime error: " + string(e.what())); set_error_message("runtime error: " + string(e.what()));
} catch (logic_error e) { } catch (logic_error e) {
setErrorCode(-1); set_error_code(-1);
setErrorMessage("logic error: " + string(e.what())); set_error_message("logic error: " + string(e.what()));
} catch (...) { } catch (...) {
setErrorCode(-2); set_error_code(-2);
setErrorMessage("unknown error"); set_error_message("unknown error");
} }
stopExecutionClock(); stop_execution_clock();
if (success) { if (success) {
setState(OperationStatus::SUCCESS); set_state(OperationStatus::SUCCESS);
} else { } else {
setState(OperationStatus::FAILED); set_state(OperationStatus::FAILED);
} }
} }
@ -124,11 +129,13 @@ bool AsyncRPCOperation_sendmany::main_impl() {
CAmount minersFee = ASYNC_RPC_OPERATION_DEFAULT_MINERS_FEE; CAmount minersFee = ASYNC_RPC_OPERATION_DEFAULT_MINERS_FEE;
// Regardless of the from address, add all taddr outputs to the raw transaction. // Regardless of the from address, add all taddr outputs to the raw transaction.
if (isfromtaddr_ && !find_utxos()) if (isfromtaddr_ && !find_utxos()) {
throw JSONRPCError(RPC_WALLET_INSUFFICIENT_FUNDS, "Insufficient funds, no UTXOs found for taddr from address."); throw JSONRPCError(RPC_WALLET_INSUFFICIENT_FUNDS, "Insufficient funds, no UTXOs found for taddr from address.");
}
if (isfromzaddr_ && !find_unspent_notes()) if (isfromzaddr_ && !find_unspent_notes()) {
throw JSONRPCError(RPC_WALLET_INSUFFICIENT_FUNDS, "Insufficient funds, no unspent notes found for zaddr from address."); throw JSONRPCError(RPC_WALLET_INSUFFICIENT_FUNDS, "Insufficient funds, no unspent notes found for zaddr from address.");
}
CAmount t_inputs_total = 0; CAmount t_inputs_total = 0;
for (SendManyInputUTXO & t : t_inputs_) { for (SendManyInputUTXO & t : t_inputs_) {
@ -163,12 +170,13 @@ bool AsyncRPCOperation_sendmany::main_impl() {
std::cout << "targetAmount: " << targetAmount << std::endl; std::cout << "targetAmount: " << targetAmount << std::endl;
#endif #endif
if (isfromtaddr_ && (t_inputs_total < targetAmount)) if (isfromtaddr_ && (t_inputs_total < targetAmount)) {
throw JSONRPCError(RPC_WALLET_INSUFFICIENT_FUNDS, strprintf("Insufficient transparent funds, have %ld, need %ld plus fee %ld", t_inputs_total, t_outputs_total, minersFee)); throw JSONRPCError(RPC_WALLET_INSUFFICIENT_FUNDS, strprintf("Insufficient transparent funds, have %ld, need %ld plus fee %ld", t_inputs_total, t_outputs_total, minersFee));
}
if (isfromzaddr_ && (z_inputs_total < targetAmount)) if (isfromzaddr_ && (z_inputs_total < targetAmount)) {
throw JSONRPCError(RPC_WALLET_INSUFFICIENT_FUNDS, strprintf("Insufficient protected funds, have %ld, need %ld plus fee %ld", z_inputs_total, t_outputs_total, minersFee)); throw JSONRPCError(RPC_WALLET_INSUFFICIENT_FUNDS, strprintf("Insufficient protected funds, have %ld, need %ld plus fee %ld", z_inputs_total, t_outputs_total, minersFee));
}
// If from address is a taddr, select UTXOs to spend // If from address is a taddr, select UTXOs to spend
CAmount selectedUTXOAmount = 0; CAmount selectedUTXOAmount = 0;
@ -177,9 +185,10 @@ bool AsyncRPCOperation_sendmany::main_impl() {
for (SendManyInputUTXO & t : t_inputs_) { for (SendManyInputUTXO & t : t_inputs_) {
selectedUTXOAmount += std::get<2>(t); selectedUTXOAmount += std::get<2>(t);
selectedTInputs.push_back(t); selectedTInputs.push_back(t);
if (selectedUTXOAmount >= targetAmount) if (selectedUTXOAmount >= targetAmount) {
break; break;
} }
}
t_inputs_ = selectedTInputs; t_inputs_ = selectedTInputs;
t_inputs_total = selectedUTXOAmount; t_inputs_total = selectedUTXOAmount;
@ -273,8 +282,9 @@ bool AsyncRPCOperation_sendmany::main_impl() {
if (hexMemo.size() > 0) { if (hexMemo.size() > 0) {
std::vector<unsigned char> rawMemo = ParseHex(hexMemo.c_str()); std::vector<unsigned char> rawMemo = ParseHex(hexMemo.c_str());
boost::array<unsigned char, ZC_MEMO_SIZE> memo = {{0x00}}; boost::array<unsigned char, ZC_MEMO_SIZE> memo = {{0x00}};
if (rawMemo.size() > ZC_MEMO_SIZE) if (rawMemo.size() > ZC_MEMO_SIZE) {
throw JSONRPCError(RPC_INVALID_PARAMETER, strprintf("Memo size of %d is too big, maximum allowed is %d", rawMemo.size(), ZC_MEMO_SIZE)); throw JSONRPCError(RPC_INVALID_PARAMETER, strprintf("Memo size of %d is too big, maximum allowed is %d", rawMemo.size(), ZC_MEMO_SIZE));
}
int lenMemo = rawMemo.size(); int lenMemo = rawMemo.size();
for (int i = 0; i < ZC_MEMO_SIZE && i < lenMemo; i++) { for (int i = 0; i < ZC_MEMO_SIZE && i < lenMemo; i++) {
@ -294,9 +304,9 @@ bool AsyncRPCOperation_sendmany::main_impl() {
// Private change will flow back to sender's zaddr, while transparent change flows to a new taddr. // Private change will flow back to sender's zaddr, while transparent change flows to a new taddr.
CAmount change = funds - fundsSpent; CAmount change = funds - fundsSpent;
if (change < 0) if (change < 0) {
throw JSONRPCError(RPC_WALLET_INSUFFICIENT_FUNDS, strprintf("Insufficient funds or internal error, spent too much leaving negative change %ld", change)); throw JSONRPCError(RPC_WALLET_INSUFFICIENT_FUNDS, strprintf("Insufficient funds or internal error, spent too much leaving negative change %ld", change));
if (change > 0) { } else if (change > 0) {
if (isfromzaddr_) { if (isfromzaddr_) {
info.vjsout.push_back(JSOutput(frompaymentaddress_, change)); info.vjsout.push_back(JSOutput(frompaymentaddress_, change));
} else if (isfromtaddr_) { } else if (isfromtaddr_) {
@ -308,8 +318,9 @@ bool AsyncRPCOperation_sendmany::main_impl() {
CReserveKey keyChange(pwalletMain); CReserveKey keyChange(pwalletMain);
CPubKey vchPubKey; CPubKey vchPubKey;
bool ret = keyChange.GetReservedKey(vchPubKey); bool ret = keyChange.GetReservedKey(vchPubKey);
if (!ret) if (!ret) {
throw JSONRPCError(RPC_WALLET_KEYPOOL_RAN_OUT, "Could not generate a taddr to use as a change address"); // should never fail, as we just unlocked throw JSONRPCError(RPC_WALLET_KEYPOOL_RAN_OUT, "Could not generate a taddr to use as a change address"); // should never fail, as we just unlocked
}
CScript scriptPubKey = GetScriptForDestination(vchPubKey.GetID()); CScript scriptPubKey = GetScriptForDestination(vchPubKey.GetID());
CTxOut out(change, scriptPubKey); CTxOut out(change, scriptPubKey);
rawTx.vout.push_back(out); rawTx.vout.push_back(out);
@ -359,7 +370,7 @@ bool AsyncRPCOperation_sendmany::main_impl() {
Object o; Object o;
o.push_back(Pair("txid", txid)); o.push_back(Pair("txid", txid));
//o.push_back(Pair("hex", signedtxn)); //o.push_back(Pair("hex", signedtxn));
setResult(Value(o)); set_result(Value(o));
} else { } else {
// Test mode does not send the transaction to the network. // Test mode does not send the transaction to the network.
@ -371,7 +382,7 @@ bool AsyncRPCOperation_sendmany::main_impl() {
o.push_back(Pair("test", 1)); o.push_back(Pair("test", 1));
o.push_back(Pair("txid", tx.GetTxid().ToString())); o.push_back(Pair("txid", tx.GetTxid().ToString()));
o.push_back(Pair("hex", signedtxn)); o.push_back(Pair("hex", signedtxn));
setResult(Value(o)); set_result(Value(o));
} }
return true; return true;
@ -387,17 +398,20 @@ bool AsyncRPCOperation_sendmany::find_utxos() {
pwalletMain->AvailableCoins(vecOutputs, false, NULL, true); pwalletMain->AvailableCoins(vecOutputs, false, NULL, true);
BOOST_FOREACH(const COutput& out, vecOutputs) { BOOST_FOREACH(const COutput& out, vecOutputs) {
if (out.nDepth < mindepth_) if (out.nDepth < mindepth_) {
continue; continue;
}
if (setAddress.size()) { if (setAddress.size()) {
CTxDestination address; CTxDestination address;
if (!ExtractDestination(out.tx->vout[out.i].scriptPubKey, address)) if (!ExtractDestination(out.tx->vout[out.i].scriptPubKey, address)) {
continue; continue;
}
if (!setAddress.count(address)) if (!setAddress.count(address)) {
continue; continue;
} }
}
// TODO: Also examine out.fSpendable ? // TODO: Also examine out.fSpendable ?
CAmount nValue = out.tx->vout[out.i].nValue; CAmount nValue = out.tx->vout[out.i].nValue;
@ -416,13 +430,15 @@ bool AsyncRPCOperation_sendmany::find_unspent_notes() {
CWalletTx wtx = p.second; CWalletTx wtx = p.second;
// Filter the transactions before checking for notes // Filter the transactions before checking for notes
if (!CheckFinalTx(wtx) || wtx.GetBlocksToMaturity() > 0 || wtx.GetDepthInMainChain() < mindepth_) if (!CheckFinalTx(wtx) || wtx.GetBlocksToMaturity() > 0 || wtx.GetDepthInMainChain() < mindepth_) {
continue; continue;
}
mapNoteData_t mapNoteData = pwalletMain->FindMyNotes(wtx); mapNoteData_t mapNoteData = pwalletMain->FindMyNotes(wtx);
if (mapNoteData.size() == 0) if (mapNoteData.size() == 0) {
continue; continue;
}
for (auto & pair : mapNoteData) { for (auto & pair : mapNoteData) {
JSOutPoint jsop = pair.first; JSOutPoint jsop = pair.first;
@ -431,8 +447,9 @@ bool AsyncRPCOperation_sendmany::find_unspent_notes() {
PaymentAddress pa = nd.address; PaymentAddress pa = nd.address;
// skip notes which belong to a different payment address in the wallet // skip notes which belong to a different payment address in the wallet
if (!(pa == frompaymentaddress_)) if (!(pa == frompaymentaddress_)) {
continue; continue;
}
int i = jsop.js; // Index into CTransaction.vjoinsplit int i = jsop.js; // Index into CTransaction.vjoinsplit
int j = jsop.n; // Index into JSDescription.ciphertexts int j = jsop.n; // Index into JSDescription.ciphertexts
@ -451,8 +468,9 @@ bool AsyncRPCOperation_sendmany::find_unspent_notes() {
uint256 nullifier = plaintext.note(frompaymentaddress_).nullifier(spendingkey_); uint256 nullifier = plaintext.note(frompaymentaddress_).nullifier(spendingkey_);
bool isSpent = pwalletMain->IsSpent(nullifier); bool isSpent = pwalletMain->IsSpent(nullifier);
if (isSpent) if (isSpent) {
continue; continue;
}
z_inputs_.push_back(SendManyInputNPT(plaintext, CAmount(plaintext.value))); z_inputs_.push_back(SendManyInputNPT(plaintext, CAmount(plaintext.value)));
@ -473,8 +491,9 @@ bool AsyncRPCOperation_sendmany::find_unspent_notes() {
} }
} }
if (z_inputs_.size() == 0) if (z_inputs_.size() == 0) {
return false; return false;
}
// sort in descending order, so big notes appear first // sort in descending order, so big notes appear first
std::sort(z_inputs_.begin(), z_inputs_.end(), [](SendManyInputNPT i, SendManyInputNPT j) -> bool { std::sort(z_inputs_.begin(), z_inputs_.end(), [](SendManyInputNPT i, SendManyInputNPT j) -> bool {
@ -496,8 +515,9 @@ Object AsyncRPCOperation_sendmany::perform_joinsplit(AsyncJoinSplitInfo & info)
// Unlock critical section // Unlock critical section
if (!(witnesses.size() == info.notes.size()) || !(info.notes.size() == info.keys.size())) if (!(witnesses.size() == info.notes.size()) || !(info.notes.size() == info.keys.size())) {
throw runtime_error("number of notes and witnesses and keys do not match"); throw runtime_error("number of notes and witnesses and keys do not match");
}
for (size_t i = 0; i < witnesses.size(); i++) { for (size_t i = 0; i < witnesses.size(); i++) {
if (!witnesses[i]) { if (!witnesses[i]) {
@ -548,8 +568,9 @@ Object AsyncRPCOperation_sendmany::perform_joinsplit(AsyncJoinSplitInfo & info)
info.vpub_old, info.vpub_old,
info.vpub_new); info.vpub_new);
if (!(jsdesc.Verify(*pzcashParams, joinSplitPubKey))) if (!(jsdesc.Verify(*pzcashParams, joinSplitPubKey))) {
throw std::runtime_error("error verifying joinsplt"); throw std::runtime_error("error verifying joinsplt");
}
mtx.vjoinsplit.push_back(jsdesc); mtx.vjoinsplit.push_back(jsdesc);
@ -563,14 +584,18 @@ Object AsyncRPCOperation_sendmany::perform_joinsplit(AsyncJoinSplitInfo & info)
dataToBeSigned.begin(), 32, dataToBeSigned.begin(), 32,
joinSplitPrivKey joinSplitPrivKey
) == 0)) ) == 0))
{
throw std::runtime_error("crypto_sign_detached failed"); throw std::runtime_error("crypto_sign_detached failed");
}
// Sanity check // Sanity check
if (!(crypto_sign_verify_detached(&mtx.joinSplitSig[0], if (!(crypto_sign_verify_detached(&mtx.joinSplitSig[0],
dataToBeSigned.begin(), 32, dataToBeSigned.begin(), 32,
mtx.joinSplitPubKey.begin() mtx.joinSplitPubKey.begin()
) == 0)) ) == 0))
{
throw std::runtime_error("crypto_sign_verify_detached failed"); throw std::runtime_error("crypto_sign_verify_detached failed");
}
CTransaction rawTx(mtx); CTransaction rawTx(mtx);
tx_ = rawTx; tx_ = rawTx;
@ -615,8 +640,9 @@ void AsyncRPCOperation_sendmany::add_taddr_outputs_to_tx() {
CAmount nAmount = std::get<1>(r); CAmount nAmount = std::get<1>(r);
CBitcoinAddress address(outputAddress); CBitcoinAddress address(outputAddress);
if (!address.IsValid()) if (!address.IsValid()) {
throw JSONRPCError(RPC_INVALID_ADDRESS_OR_KEY, "Invalid output address, not a valid taddr."); throw JSONRPCError(RPC_INVALID_ADDRESS_OR_KEY, "Invalid output address, not a valid taddr.");
}
CScript scriptPubKey = GetScriptForDestination(address.Get()); CScript scriptPubKey = GetScriptForDestination(address.Get());