Skip to content

Commit 6aaf424

Browse files
andreacampifacebook-github-bot
andreacampi
authored and
facebook-github-bot
committed
Move the blobstore fetch of contents out of the Edenapi call
Summary: ## This stack I want to be able to split loading from blobstore from sending via Edenapi. The main motivation right now is benchmarking, but we might be able to optimize further. ## This diff With this we'll be more granular in our benchmarking. We can also look at more optimizations. Reviewed By: RajivTS Differential Revision: D72063616 fbshipit-source-id: f145d86fb6888c2c815d5d707127194ca82348fd
1 parent 229ad04 commit 6aaf424

File tree

6 files changed

+53
-42
lines changed

6 files changed

+53
-42
lines changed

‎eden/mononoke/modern_sync/src/sender/edenapi.rs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,13 +7,14 @@
77

88
use anyhow::Result;
99
use async_trait::async_trait;
10+
use edenapi_types::AnyFileContentId;
1011
use mercurial_types::blobs::HgBlobChangeset;
1112
use mercurial_types::HgChangesetId;
1213
use mercurial_types::HgFileNodeId;
1314
use mercurial_types::HgManifestId;
15+
use minibytes::Bytes;
1416
use mononoke_types::BonsaiChangeset;
1517
use mononoke_types::ChangesetId;
16-
use mononoke_types::ContentId;
1718

1819
mod default;
1920
mod filter;
@@ -29,7 +30,7 @@ pub(crate) use retry::RetryEdenapiSender;
2930

3031
#[async_trait]
3132
pub(crate) trait EdenapiSender {
32-
async fn upload_contents(&self, contents: Vec<ContentId>) -> Result<()>;
33+
async fn upload_contents(&self, contents: Vec<(AnyFileContentId, Bytes)>) -> Result<()>;
3334

3435
async fn upload_trees(&self, trees: Vec<HgManifestId>) -> Result<()>;
3536

‎eden/mononoke/modern_sync/src/sender/edenapi/default.rs‎

Lines changed: 4 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -11,13 +11,10 @@ use std::time::Duration;
1111

1212
use anyhow::anyhow;
1313
use anyhow::ensure;
14-
use anyhow::Error;
1514
use anyhow::Result;
1615
use async_trait::async_trait;
17-
use bytes::BytesMut;
1816
use clientinfo::ClientEntryPoint;
1917
use clientinfo::ClientInfo;
20-
use cloned::cloned;
2118
use context::CoreContext;
2219
use edenapi::api::UploadLookupPolicy;
2320
use edenapi::paths;
@@ -40,10 +37,10 @@ use mercurial_types::blobs::HgBlobChangeset;
4037
use mercurial_types::HgChangesetId;
4138
use mercurial_types::HgFileNodeId;
4239
use mercurial_types::HgManifestId;
40+
use minibytes::Bytes;
4341
use mononoke_app::args::TLSArgs;
4442
use mononoke_types::BonsaiChangeset;
4543
use mononoke_types::ChangesetId;
46-
use mononoke_types::ContentId;
4744
use repo_blobstore::RepoBlobstore;
4845
use slog::info;
4946
use slog::Logger;
@@ -120,33 +117,13 @@ impl DefaultEdenapiSender {
120117

121118
#[async_trait]
122119
impl EdenapiSender for DefaultEdenapiSender {
123-
async fn upload_contents(&self, contents: Vec<ContentId>) -> Result<()> {
124-
let repo_blobstore = self.repo_blobstore.clone();
120+
async fn upload_contents(&self, contents: Vec<(AnyFileContentId, Bytes)>) -> Result<()> {
125121
let ctx = self.ctx.clone();
126-
let len = contents.len();
127-
let full_items = stream::iter(contents)
128-
.map(|id| {
129-
cloned!(ctx, repo_blobstore);
130-
async move {
131-
let bytes = filestore::fetch(repo_blobstore, ctx, &id.into())
132-
.await?
133-
.ok_or(anyhow!("Content is not found (which should never happen"))?
134-
.try_collect::<BytesMut>()
135-
.await?;
136-
Ok::<_, Error>((
137-
AnyFileContentId::ContentId(id.into()),
138-
bytes.freeze().into(),
139-
))
140-
}
141-
})
142-
.buffer_unordered(len)
143-
.try_collect::<Vec<(AnyFileContentId, minibytes::Bytes)>>()
144-
.await?;
145122

146-
let expected_responses = full_items.len();
123+
let expected_responses = contents.len();
147124
let response = self
148125
.client()?
149-
.process_files_upload(full_items, None, None, UploadLookupPolicy::SkipLookup)
126+
.process_files_upload(contents, None, None, UploadLookupPolicy::SkipLookup)
150127
.await?;
151128

152129
let ids = response

‎eden/mononoke/modern_sync/src/sender/edenapi/filter.rs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -11,13 +11,14 @@ use std::sync::Arc;
1111
use anyhow::Ok;
1212
use anyhow::Result;
1313
use async_trait::async_trait;
14+
use edenapi_types::AnyFileContentId;
1415
use mercurial_types::blobs::HgBlobChangeset;
1516
use mercurial_types::HgChangesetId;
1617
use mercurial_types::HgFileNodeId;
1718
use mercurial_types::HgManifestId;
19+
use minibytes::Bytes;
1820
use mononoke_types::BonsaiChangeset;
1921
use mononoke_types::ChangesetId;
20-
use mononoke_types::ContentId;
2122

2223
use crate::sender::edenapi::EdenapiSender;
2324

@@ -48,7 +49,7 @@ impl FilterEdenapiSender {
4849

4950
#[async_trait]
5051
impl EdenapiSender for FilterEdenapiSender {
51-
async fn upload_contents(&self, contents: Vec<ContentId>) -> Result<()> {
52+
async fn upload_contents(&self, contents: Vec<(AnyFileContentId, Bytes)>) -> Result<()> {
5253
if self
5354
.allowed
5455
.get(&MethodFilter::UploadContents)

‎eden/mononoke/modern_sync/src/sender/edenapi/noop.rs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -7,13 +7,14 @@
77

88
use anyhow::Result;
99
use async_trait::async_trait;
10+
use edenapi_types::AnyFileContentId;
1011
use mercurial_types::blobs::HgBlobChangeset;
1112
use mercurial_types::HgChangesetId;
1213
use mercurial_types::HgFileNodeId;
1314
use mercurial_types::HgManifestId;
15+
use minibytes::Bytes;
1416
use mononoke_types::BonsaiChangeset;
1517
use mononoke_types::ChangesetId;
16-
use mononoke_types::ContentId;
1718

1819
use crate::sender::edenapi::EdenapiSender;
1920

@@ -22,7 +23,7 @@ pub struct NoopEdenapiSender {}
2223

2324
#[async_trait]
2425
impl EdenapiSender for NoopEdenapiSender {
25-
async fn upload_contents(&self, _contents: Vec<ContentId>) -> Result<()> {
26+
async fn upload_contents(&self, _contents: Vec<(AnyFileContentId, Bytes)>) -> Result<()> {
2627
Ok(())
2728
}
2829

‎eden/mononoke/modern_sync/src/sender/edenapi/retry.rs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,15 +10,16 @@ use std::time::Duration;
1010

1111
use anyhow::Result;
1212
use async_trait::async_trait;
13+
use edenapi_types::AnyFileContentId;
1314
use futures::future::BoxFuture;
1415
use futures::FutureExt;
1516
use mercurial_types::blobs::HgBlobChangeset;
1617
use mercurial_types::HgChangesetId;
1718
use mercurial_types::HgFileNodeId;
1819
use mercurial_types::HgManifestId;
20+
use minibytes::Bytes;
1921
use mononoke_types::BonsaiChangeset;
2022
use mononoke_types::ChangesetId;
21-
use mononoke_types::ContentId;
2223
use slog::warn;
2324
use slog::Logger;
2425

@@ -47,7 +48,7 @@ impl RetryEdenapiSender {
4748

4849
#[async_trait]
4950
impl EdenapiSender for RetryEdenapiSender {
50-
async fn upload_contents(&self, contents: Vec<ContentId>) -> Result<()> {
51+
async fn upload_contents(&self, contents: Vec<(AnyFileContentId, Bytes)>) -> Result<()> {
5152
self.with_retry(|this| this.inner.upload_contents(contents.clone()).boxed())
5253
.await
5354
}

‎eden/mononoke/modern_sync/src/sender/manager/content.rs‎

Lines changed: 37 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -10,10 +10,18 @@ use std::sync::atomic::AtomicBool;
1010
use std::sync::atomic::Ordering;
1111
use std::sync::Arc;
1212

13+
use anyhow::anyhow;
14+
use anyhow::Error;
1315
use anyhow::Result;
16+
use bytes::BytesMut;
1417
use bytesize::ByteSize;
18+
use cloned::cloned;
1519
use context::CoreContext;
20+
use edenapi_types::AnyFileContentId;
1621
use futures::channel::oneshot;
22+
use futures::stream;
23+
use futures::StreamExt;
24+
use futures::TryStreamExt;
1725
use mononoke_macros::mononoke;
1826
use mononoke_types::ContentId;
1927
use repo_blobstore::RepoBlobstore;
@@ -63,21 +71,41 @@ impl ContentManager {
6371
}
6472

6573
async fn flush_batch(
66-
_ctx: CoreContext,
74+
ctx: CoreContext,
75+
repo_blobstore: RepoBlobstore,
6776
content_es: &Arc<dyn EdenapiSender + Send + Sync>,
6877
current_batch: &mut Vec<ContentId>,
6978
current_batch_size: u64,
7079
pending_messages: &mut VecDeque<oneshot::Sender<Result<(), anyhow::Error>>>,
7180
reponame: String,
7281
logger: &Logger,
7382
) -> Result<(), anyhow::Error> {
74-
let current_batch_len = current_batch.len() as i64;
83+
let current_batch_len = current_batch.len();
7584
let start = std::time::Instant::now();
85+
7686
if current_batch_len > 0 {
77-
if let Err(e) = content_es
78-
.upload_contents(std::mem::take(current_batch))
79-
.await
80-
{
87+
let contents = std::mem::take(current_batch);
88+
89+
let full_items = stream::iter(contents)
90+
.map(|id| {
91+
cloned!(ctx, repo_blobstore);
92+
async move {
93+
let bytes = filestore::fetch(repo_blobstore, ctx, &id.into())
94+
.await?
95+
.ok_or(anyhow!("Content is not found (which should never happen"))?
96+
.try_collect::<BytesMut>()
97+
.await?;
98+
Ok::<_, Error>((
99+
AnyFileContentId::ContentId(id.into()),
100+
bytes.freeze().into(),
101+
))
102+
}
103+
})
104+
.buffer_unordered(current_batch_len)
105+
.try_collect::<Vec<(AnyFileContentId, minibytes::Bytes)>>()
106+
.await?;
107+
108+
if let Err(e) = content_es.upload_contents(full_items).await {
81109
error!(logger, "Error processing content: {:?}", e);
82110
return Err(e);
83111
} else {
@@ -91,7 +119,7 @@ impl ContentManager {
91119

92120
let elapsed = start.elapsed().as_millis() / current_batch_len as u128;
93121
STATS::content_upload_time_s.add_value(elapsed as i64, (reponame.clone(),));
94-
STATS::synced_contents.add_value(current_batch_len, (reponame.clone(),));
122+
STATS::synced_contents.add_value(current_batch_len as i64, (reponame.clone(),));
95123
}
96124
}
97125

@@ -144,6 +172,7 @@ impl Manager for ContentManager {
144172
if current_batch_size >= MAX_BLOB_BYTES || current_batch.len() >= MAX_CONTENT_BATCH_SIZE {
145173
if let Err(e) = ContentManager::flush_batch(
146174
ctx.clone(),
175+
self.repo_blobstore.clone(),
147176
&content_es,
148177
&mut current_batch,
149178
current_batch_size,
@@ -161,6 +190,7 @@ impl Manager for ContentManager {
161190
if current_batch_size > 0 || !pending_messages.is_empty() {
162191
if let Err(e) = ContentManager::flush_batch(
163192
ctx.clone(),
193+
self.repo_blobstore.clone(),
164194
&content_es,
165195
&mut current_batch,
166196
current_batch_size,

0 commit comments

Comments
 (0)