Page MenuHomePhabricator
Differential D3534

[services] Backup - Print out errors in reactors
ClosedPublic
Actions

Authored by • karol on Mar 29 2022, 4:39 AM.
Tags
None
Referenced Files
F4948800: D3534.diff
Thu, Mar 20, 12:17 AM
Unknown Object (File)
Sun, Mar 16, 12:39 PM
Unknown Object (File)
Sun, Mar 16, 12:39 PM
Unknown Object (File)
Sun, Mar 16, 12:39 PM
Unknown Object (File)
Sun, Mar 16, 12:36 PM
Unknown Object (File)
Thu, Mar 13, 6:10 AM
Unknown Object (File)
Tue, Mar 4, 7:51 AM
Unknown Object (File)
Thu, Feb 27, 2:51 AM
View All Files
Subscribers
• abosh
• adrian
ashoat
atul
• benschac
tomek

Details

Reviewers
• max
tomek
varun
• jim
Commits
rCOMM2daf41aac622: [services] Backup - Print out errors in reactors
Summary

Depends on D3533

I thought we may want to print out error messages in reactors to the stdout in services - for now, they're not visible.

Test Plan
cd services
yarn run-backup-service

Diff Detail

Repository
rCOMM Comm
Lint
Lint Not Applicable
Unit
Tests Not Applicable

Revision Contents
Changeset List

PathSize
services/
backup/
docker-server/
contents/
server/
src/
Reactors/
client/
base-reactors/
3 lines
3 lines
3 lines
server/
base-reactors/
3 lines
3 lines
3 lines
blob/
src/
Reactors/
server/
base-reactors/
3 lines
3 lines

Diff 11152

View Options

services/backup/docker-server/contents/server/src/Reactors/client/base-reactors/ClientBidiReactorBase.h

  • This file was copied to services/backup/src/Reactors/client/blob/BlobPutClientReactor.cpp.
#pragma once #pragma once
#include "ClientBidiReactorBase.h"
#include "Constants.h" #include "Constants.h"
#include "../_generated/blob.grpc.pb.h" #include "../_generated/blob.grpc.pb.h"
#include "../_generated/blob.pb.h" #include "../_generated/blob.pb.h"
#include "ClientBidiReactorBase.h"
#include <folly/MPMCQueue.h> #include <folly/MPMCQueue.h>
#include <grpcpp/grpcpp.h> #include <grpcpp/grpcpp.h>
#include <condition_variable> #include <condition_variable>
#include <iostream>
#include <memory> #include <memory>
#include <string> #include <string>
namespace comm { namespace comm {
namespace network { namespace network {
namespace reactor { namespace reactor {
class BlobPutClientReactor class BlobPutClientReactor
Show All 22 Linespublic:
void scheduleSendingDataChunk(std::unique_ptr<std::string> dataChunk); void scheduleSendingDataChunk(std::unique_ptr<std::string> dataChunk);
std::unique_ptr<grpc::Status> prepareRequest( std::unique_ptr<grpc::Status> prepareRequest(
blob::PutRequest &request, blob::PutRequest &request,
std::shared_ptr<blob::PutResponse> previousResponse) override; std::shared_ptr<blob::PutResponse> previousResponse) override;
void doneCallback() override; void doneCallback() override;
grpc::Status getStatus() const; grpc::Status getStatus() const;
}; };
BlobPutClientReactor::BlobPutClientReactor(
const std::string &holder,
const std::string &hash,
std::condition_variable *terminationNotifier)
: holder(holder),
hash(hash),
dataChunks(folly::MPMCQueue<std::string>(100)),
terminationNotifier(terminationNotifier) {
}
void BlobPutClientReactor::scheduleSendingDataChunk(
std::unique_ptr<std::string> dataChunk) {
if (!this->dataChunks.write(std::move(*dataChunk))) {
throw std::runtime_error(
"Error scheduling sending a data chunk to send to the blob service");
}
}
std::unique_ptr<grpc::Status> BlobPutClientReactor::prepareRequest(
blob::PutRequest &request,
std::shared_ptr<blob::PutResponse> previousResponse) {
if (this->state == State::SEND_HOLDER) {
this->request.set_holder(this->holder);
this->state = State::SEND_HASH;
return nullptr;
}
if (this->state == State::SEND_HASH) {
request.set_blobhash(this->hash);
this->state = State::SEND_CHUNKS;
return nullptr;
}
if (previousResponse->dataexists()) {
return std::make_unique<grpc::Status>(grpc::Status::OK);
}
std::string dataChunk;
this->dataChunks.blockingRead(dataChunk);
if (dataChunk.empty()) {
return std::make_unique<grpc::Status>(grpc::Status::OK);
}
request.set_datachunk(dataChunk);
return nullptr;
}
void BlobPutClientReactor::doneCallback() {
this->terminationNotifier->notify_one();
}
grpc::Status BlobPutClientReactor::getStatus() const {
return this->status;
}
} // namespace reactor } // namespace reactor
} // namespace network } // namespace network
} // namespace comm } // namespace comm
View Options

services/backup/docker-server/contents/server/src/Reactors/client/base-reactors/ClientReadReactorBase.h

  • This file was copied to services/backup/src/Reactors/client/blob/BlobPutClientReactor.cpp.
#pragma once #pragma once
#include "ClientBidiReactorBase.h"
#include "Constants.h" #include "Constants.h"
#include "../_generated/blob.grpc.pb.h" #include "../_generated/blob.grpc.pb.h"
#include "../_generated/blob.pb.h" #include "../_generated/blob.pb.h"
#include "ClientBidiReactorBase.h"
#include <folly/MPMCQueue.h> #include <folly/MPMCQueue.h>
#include <grpcpp/grpcpp.h> #include <grpcpp/grpcpp.h>
#include <condition_variable> #include <condition_variable>
#include <iostream>
#include <memory> #include <memory>
#include <string> #include <string>
namespace comm { namespace comm {
namespace network { namespace network {
namespace reactor { namespace reactor {
class BlobPutClientReactor class BlobPutClientReactor
Show All 22 Linespublic:
void scheduleSendingDataChunk(std::unique_ptr<std::string> dataChunk); void scheduleSendingDataChunk(std::unique_ptr<std::string> dataChunk);
std::unique_ptr<grpc::Status> prepareRequest( std::unique_ptr<grpc::Status> prepareRequest(
blob::PutRequest &request, blob::PutRequest &request,
std::shared_ptr<blob::PutResponse> previousResponse) override; std::shared_ptr<blob::PutResponse> previousResponse) override;
void doneCallback() override; void doneCallback() override;
grpc::Status getStatus() const; grpc::Status getStatus() const;
}; };
BlobPutClientReactor::BlobPutClientReactor(
const std::string &holder,
const std::string &hash,
std::condition_variable *terminationNotifier)
: holder(holder),
hash(hash),
dataChunks(folly::MPMCQueue<std::string>(100)),
terminationNotifier(terminationNotifier) {
}
void BlobPutClientReactor::scheduleSendingDataChunk(
std::unique_ptr<std::string> dataChunk) {
if (!this->dataChunks.write(std::move(*dataChunk))) {
throw std::runtime_error(
"Error scheduling sending a data chunk to send to the blob service");
}
}
std::unique_ptr<grpc::Status> BlobPutClientReactor::prepareRequest(
blob::PutRequest &request,
std::shared_ptr<blob::PutResponse> previousResponse) {
if (this->state == State::SEND_HOLDER) {
this->request.set_holder(this->holder);
this->state = State::SEND_HASH;
return nullptr;
}
if (this->state == State::SEND_HASH) {
request.set_blobhash(this->hash);
this->state = State::SEND_CHUNKS;
return nullptr;
}
if (previousResponse->dataexists()) {
return std::make_unique<grpc::Status>(grpc::Status::OK);
}
std::string dataChunk;
this->dataChunks.blockingRead(dataChunk);
if (dataChunk.empty()) {
return std::make_unique<grpc::Status>(grpc::Status::OK);
}
request.set_datachunk(dataChunk);
return nullptr;
}
void BlobPutClientReactor::doneCallback() {
this->terminationNotifier->notify_one();
}
grpc::Status BlobPutClientReactor::getStatus() const {
return this->status;
}
} // namespace reactor } // namespace reactor
} // namespace network } // namespace network
} // namespace comm } // namespace comm
View Options

services/backup/docker-server/contents/server/src/Reactors/client/base-reactors/ClientWriteReactorBase.h

  • This file was copied to services/backup/src/Reactors/client/blob/BlobPutClientReactor.cpp.
#pragma once #pragma once
#include "ClientBidiReactorBase.h"
#include "Constants.h" #include "Constants.h"
#include "../_generated/blob.grpc.pb.h" #include "../_generated/blob.grpc.pb.h"
#include "../_generated/blob.pb.h" #include "../_generated/blob.pb.h"
#include "ClientBidiReactorBase.h"
#include <folly/MPMCQueue.h> #include <folly/MPMCQueue.h>
#include <grpcpp/grpcpp.h> #include <grpcpp/grpcpp.h>
#include <condition_variable> #include <condition_variable>
#include <iostream>
#include <memory> #include <memory>
#include <string> #include <string>
namespace comm { namespace comm {
namespace network { namespace network {
namespace reactor { namespace reactor {
class BlobPutClientReactor class BlobPutClientReactor
Show All 22 Linespublic:
void scheduleSendingDataChunk(std::unique_ptr<std::string> dataChunk); void scheduleSendingDataChunk(std::unique_ptr<std::string> dataChunk);
std::unique_ptr<grpc::Status> prepareRequest( std::unique_ptr<grpc::Status> prepareRequest(
blob::PutRequest &request, blob::PutRequest &request,
std::shared_ptr<blob::PutResponse> previousResponse) override; std::shared_ptr<blob::PutResponse> previousResponse) override;
void doneCallback() override; void doneCallback() override;
grpc::Status getStatus() const; grpc::Status getStatus() const;
}; };
BlobPutClientReactor::BlobPutClientReactor(
const std::string &holder,
const std::string &hash,
std::condition_variable *terminationNotifier)
: holder(holder),
hash(hash),
dataChunks(folly::MPMCQueue<std::string>(100)),
terminationNotifier(terminationNotifier) {
}
void BlobPutClientReactor::scheduleSendingDataChunk(
std::unique_ptr<std::string> dataChunk) {
if (!this->dataChunks.write(std::move(*dataChunk))) {
throw std::runtime_error(
"Error scheduling sending a data chunk to send to the blob service");
}
}
std::unique_ptr<grpc::Status> BlobPutClientReactor::prepareRequest(
blob::PutRequest &request,
std::shared_ptr<blob::PutResponse> previousResponse) {
if (this->state == State::SEND_HOLDER) {
this->request.set_holder(this->holder);
this->state = State::SEND_HASH;
return nullptr;
}
if (this->state == State::SEND_HASH) {
request.set_blobhash(this->hash);
this->state = State::SEND_CHUNKS;
return nullptr;
}
if (previousResponse->dataexists()) {
return std::make_unique<grpc::Status>(grpc::Status::OK);
}
std::string dataChunk;
this->dataChunks.blockingRead(dataChunk);
if (dataChunk.empty()) {
return std::make_unique<grpc::Status>(grpc::Status::OK);
}
request.set_datachunk(dataChunk);
return nullptr;
}
void BlobPutClientReactor::doneCallback() {
this->terminationNotifier->notify_one();
}
grpc::Status BlobPutClientReactor::getStatus() const {
return this->status;
}
} // namespace reactor } // namespace reactor
} // namespace network } // namespace network
} // namespace comm } // namespace comm
View Options

services/backup/docker-server/contents/server/src/Reactors/server/base-reactors/ServerBidiReactorBase.h

  • This file was copied to services/backup/src/Reactors/client/blob/BlobPutClientReactor.cpp.
#pragma once #pragma once
#include "ClientBidiReactorBase.h"
#include "Constants.h" #include "Constants.h"
#include "../_generated/blob.grpc.pb.h" #include "../_generated/blob.grpc.pb.h"
#include "../_generated/blob.pb.h" #include "../_generated/blob.pb.h"
#include "ClientBidiReactorBase.h"
#include <folly/MPMCQueue.h> #include <folly/MPMCQueue.h>
#include <grpcpp/grpcpp.h> #include <grpcpp/grpcpp.h>
#include <condition_variable> #include <condition_variable>
#include <iostream>
#include <memory> #include <memory>
#include <string> #include <string>
namespace comm { namespace comm {
namespace network { namespace network {
namespace reactor { namespace reactor {
class BlobPutClientReactor class BlobPutClientReactor
Show All 22 Linespublic:
void scheduleSendingDataChunk(std::unique_ptr<std::string> dataChunk); void scheduleSendingDataChunk(std::unique_ptr<std::string> dataChunk);
std::unique_ptr<grpc::Status> prepareRequest( std::unique_ptr<grpc::Status> prepareRequest(
blob::PutRequest &request, blob::PutRequest &request,
std::shared_ptr<blob::PutResponse> previousResponse) override; std::shared_ptr<blob::PutResponse> previousResponse) override;
void doneCallback() override; void doneCallback() override;
grpc::Status getStatus() const; grpc::Status getStatus() const;
}; };
BlobPutClientReactor::BlobPutClientReactor(
const std::string &holder,
const std::string &hash,
std::condition_variable *terminationNotifier)
: holder(holder),
hash(hash),
dataChunks(folly::MPMCQueue<std::string>(100)),
terminationNotifier(terminationNotifier) {
}
void BlobPutClientReactor::scheduleSendingDataChunk(
std::unique_ptr<std::string> dataChunk) {
if (!this->dataChunks.write(std::move(*dataChunk))) {
throw std::runtime_error(
"Error scheduling sending a data chunk to send to the blob service");
}
}
std::unique_ptr<grpc::Status> BlobPutClientReactor::prepareRequest(
blob::PutRequest &request,
std::shared_ptr<blob::PutResponse> previousResponse) {
if (this->state == State::SEND_HOLDER) {
this->request.set_holder(this->holder);
this->state = State::SEND_HASH;
return nullptr;
}
if (this->state == State::SEND_HASH) {
request.set_blobhash(this->hash);
this->state = State::SEND_CHUNKS;
return nullptr;
}
if (previousResponse->dataexists()) {
return std::make_unique<grpc::Status>(grpc::Status::OK);
}
std::string dataChunk;
this->dataChunks.blockingRead(dataChunk);
if (dataChunk.empty()) {
return std::make_unique<grpc::Status>(grpc::Status::OK);
}
request.set_datachunk(dataChunk);
return nullptr;
}
void BlobPutClientReactor::doneCallback() {
this->terminationNotifier->notify_one();
}
grpc::Status BlobPutClientReactor::getStatus() const {
return this->status;
}
} // namespace reactor } // namespace reactor
} // namespace network } // namespace network
} // namespace comm } // namespace comm
View Options

services/backup/docker-server/contents/server/src/Reactors/server/base-reactors/ServerReadReactorBase.h

Loading...
View Options

services/backup/docker-server/contents/server/src/Reactors/server/base-reactors/ServerWriteReactorBase.h

Loading...
View Options

services/blob/src/Reactors/server/base-reactors/ServerBidiReactorBase.h

Loading...
View Options

services/blob/src/Reactors/server/base-reactors/ServerReadReactorBase.h

Loading...