daqbuf/daqbuffer
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add AggKind variant for N bins in X-dim

Browse Source
This commit is contained in:
Dominik Werder
2021-05-19 07:28:44 +02:00
parent d4d76d97da
commit 3b0404f2ba
3 changed files with 40 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
disk/src/binned.rs
View File

@@ -10,7 +10,7 @@ use err::Error;
use futures_core::Stream;
use futures_util::StreamExt;
use netpod::log::*;
use netpod::{BinnedRange, NodeConfigCached, PerfOpts, PreBinnedPatchIterator, PreBinnedPatchRange};
use netpod::{AggKind, BinnedRange, NodeConfigCached, PerfOpts, PreBinnedPatchIterator, PreBinnedPatchRange};
use num_traits::Zero;
use serde::{Deserialize, Serialize, Serializer};
use std::pin::Pin;
@@ -100,12 +100,15 @@ pub async fn binned_bytes_for_http(
query: &BinnedQuery,
) -> Result<BinnedStreamBox, Error> {
// TODO must decide here already which AggKind so that I can call into the generic code.
todo::todo;
let res = binned_stream(node_config, query).await?;
let ret = BinnedBytesForHttpStream::new(res.binned_stream);
Ok(Box::pin(ret))
//todo::todo;
match query.agg_kind() {
AggKind::DimXBins1 => {
let res = binned_stream(node_config, query).await?;
let ret = BinnedBytesForHttpStream::new(res.binned_stream);
Ok(Box::pin(ret))
}
AggKind::DimXBinsN(_) => err::todoval(),
}
}
pub type BinnedBytesForHttpStreamFrame = <BinnedStreamFromPreBinnedPatches as Stream>::Item;

8
disk/src/cache.rs
View File

@@ -186,7 +186,7 @@ impl PreBinnedQuery {
patch: PreBinnedPatchCoord::new(bin_t_len, patch_t_len, patch_ix),
agg_kind: params
.get("agg_kind")
.map_or(&format!("{:?}", AggKind::DimXBins1), |k| k)
.map_or(&format!("{}", AggKind::DimXBins1), |k| k)
.parse()
.map_err(|e| Error::with_msg(format!("can not parse agg_kind {:?}", e)))?,
channel: channel_from_params(&params)?,
@@ -198,7 +198,7 @@ impl PreBinnedQuery {
pub fn make_query_string(&self) -> String {
format!(
"{}&channel_backend={}&channel_name={}&agg_kind={:?}&cache_usage={}&disk_stats_every_kb={}",
"{}&channel_backend={}&channel_name={}&agg_kind={}&cache_usage={}&disk_stats_every_kb={}",
self.patch.to_url_params_strings(),
self.channel.backend,
self.channel.name,
@@ -452,7 +452,7 @@ impl CacheFileDesc {
h.update(b"V000");
h.update(self.channel.backend.as_bytes());
h.update(self.channel.name.as_bytes());
h.update(format!("{:?}", self.agg_kind).as_bytes());
h.update(format!("{}", self.agg_kind).as_bytes());
h.update(&self.patch.spec().bin_t_len().to_le_bytes());
h.update(&self.patch.spec().patch_t_len().to_le_bytes());
h.update(&self.patch.ix().to_le_bytes());
@@ -481,7 +481,7 @@ impl CacheFileDesc {
.join(&hc[0..3])
.join(&hc[3..6])
.join(&self.channel.name)
.join(format!("{:?}", self.agg_kind))
.join(format!("{}", self.agg_kind))
.join(format!(
"{:010}-{:010}",
self.patch.spec().bin_t_len() / SEC,