diff --git a/ecstore/src/cache_value/metacache_set.rs b/ecstore/src/cache_value/metacache_set.rs index c3e75d093..1e32b4391 100644 --- a/ecstore/src/cache_value/metacache_set.rs +++ b/ecstore/src/cache_value/metacache_set.rs @@ -7,7 +7,7 @@ use common::error::{Error, Result}; use futures::future::join_all; use std::{future::Future, pin::Pin, sync::Arc}; use tokio::{spawn, sync::broadcast::Receiver as B_Receiver}; -use tracing::{error, info, warn}; +use tracing::error; pub type AgreedFn = Box Pin + Send>> + Send + 'static>; pub type PartialFn = Box]) -> Pin + Send>> + Send + 'static>; @@ -54,7 +54,6 @@ impl Clone for ListPathRawOptions { pub async fn list_path_raw(mut rx: B_Receiver, opts: ListPathRawOptions) -> Result<()> { // println!("list_path_raw {},{}", &opts.bucket, &opts.path); if opts.disks.is_empty() { - info!("list_path_raw 0 drives provided"); return Err(Error::from_string("list_path_raw: 0 drives provided")); } @@ -214,16 +213,12 @@ pub async fn list_path_raw(mut rx: B_Receiver, opts: ListPathRawOptions) - // If only the name matches we didn't agree, but add it for resolution. if entry.name == current.name { top_entries[i] = Some(entry); - continue; } // We got different entries if entry.name > current.name { continue; } - // We got a new, better current. - // Clear existing entries. - top_entries = vec![None; top_entries.len()]; for item in top_entries.iter_mut().take(i) { *item = None; @@ -270,7 +265,6 @@ pub async fn list_path_raw(mut rx: B_Receiver, opts: ListPathRawOptions) - } } - warn!("list_path_raw: all at eof or error"); break; } @@ -279,7 +273,6 @@ pub async fn list_path_raw(mut rx: B_Receiver, opts: ListPathRawOptions) - let _ = r.skip(1).await; } - warn!("list_path_raw: agree == readers.len() {} ", ¤t.name); if let Some(agreed_fn) = opts.agreed.as_ref() { agreed_fn(current).await; } @@ -293,7 +286,6 @@ pub async fn list_path_raw(mut rx: B_Receiver, opts: ListPathRawOptions) - } } - warn!("list_path_raw: {} entries", top_entries.len()); if let Some(partial_fn) = opts.partial.as_ref() { partial_fn(MetaCacheEntries(top_entries), &errs).await; } diff --git a/ecstore/src/pools.rs b/ecstore/src/pools.rs index c151b0627..8a0761006 100644 --- a/ecstore/src/pools.rs +++ b/ecstore/src/pools.rs @@ -31,6 +31,7 @@ use std::io::{Cursor, Write}; use std::path::PathBuf; use std::sync::Arc; use time::{Duration, OffsetDateTime}; +use tokio::io::AsyncReadExt; use tokio::sync::broadcast::Receiver as B_Receiver; use tracing::{error, info, warn}; @@ -910,7 +911,11 @@ impl ECStore { let wk = wk.clone(); let set = set.clone(); let rcfg = rcfg.clone(); - Box::pin(async move { this.decommission_entry(idx, entry, bucket, set, wk, rcfg).await }) + + Box::pin(async move { + wk.take().await; + this.decommission_entry(idx, entry, bucket, set, wk, rcfg).await + }) } }); @@ -918,6 +923,7 @@ impl ECStore { let mut rx = rx.resubscribe(); let bi = bi.clone(); let set_id = set_idx; + let wk_clone = wk.clone(); tokio::spawn(async move { loop { if rx.try_recv().is_ok() { @@ -945,6 +951,8 @@ impl ECStore { } } } + + wk_clone.give().await; }); } @@ -1166,7 +1174,7 @@ impl ECStore { #[tracing::instrument(skip(self, rd))] async fn decommission_object(self: Arc, pool_idx: usize, bucket: String, rd: GetObjectReader) -> Result<()> { - warn!("decommission_object: {} {}", &bucket, &rd.object_info.name); + warn!("decommission_object: start {} {}", &bucket, &rd.object_info.name); let object_info = rd.object_info.clone(); // TODO: check : use size or actual_size ? @@ -1194,8 +1202,6 @@ impl ECStore { } }; - // TODO: defer abort_multipart_upload - defer!(|| async { if let Err(err) = self .abort_multipart_upload(&bucket, &object_info.name, &res.upload_id, &ObjectOptions::default()) @@ -1210,9 +1216,13 @@ impl ECStore { let mut reader = rd.stream; for (i, part) in object_info.parts.iter().enumerate() { - // 每次从reader中读取一个part上传 + let mut chunk = vec![0u8; part.size]; - let mut data = PutObjReader::new(reader, part.size); + reader.read_exact(&mut chunk).await?; + + // 每次从reader中读取一个part上传 + let rd = Box::new(Cursor::new(chunk)); + let mut data = PutObjReader::new(rd, part.size); let pi = match self .put_object_part( @@ -1230,18 +1240,17 @@ impl ECStore { { Ok(pi) => pi, Err(err) => { - error!("decommission_object: put_object_part err {:?}", &err); + error!("decommission_object: put_object_part {} err {:?}", i, &err); return Err(err); } }; + warn!("decommission_object: put_object_part {} done {} {}", i, &bucket, &object_info.name); + parts[i] = CompletePart { part_num: pi.part_num, e_tag: pi.etag, }; - - // 把reader所有权拿回来? - reader = data.stream; } if let Err(err) = self @@ -1262,6 +1271,7 @@ impl ECStore { return Err(err); } + warn!("decommission_object: complete_multipart_upload done {} {}", &bucket, &object_info.name); return Ok(()); } @@ -1289,6 +1299,7 @@ impl ECStore { return Err(err); } + warn!("decommission_object: put_object done {} {}", &bucket, &object_info.name); Ok(()) } } @@ -1319,6 +1330,8 @@ impl SetDisks { ..Default::default() }; + let cb1 = cb_func.clone(); + list_path_raw( rx, ListPathRawOptions { @@ -1327,6 +1340,7 @@ impl SetDisks { path: bucket_info.prefix.clone(), recursice: true, min_disks: listing_quorum, + agreed: Some(Box::new(move |entry: MetaCacheEntry| Box::pin(cb1(entry)))), partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option]| { let resolver = resolver.clone(); let cb_func = cb_func.clone(); diff --git a/ecstore/src/rebalance.rs b/ecstore/src/rebalance.rs index d6ec57113..97255caf8 100644 --- a/ecstore/src/rebalance.rs +++ b/ecstore/src/rebalance.rs @@ -370,7 +370,7 @@ impl ECStore { let rebalance_meta = self.rebalance_meta.read().await; if let Some(meta) = rebalance_meta.as_ref() { if let Some(pool_stat) = meta.pool_stats.get(pool_index) { - if pool_stat.info.status != RebalStatus::Completed || !pool_stat.participating { + if pool_stat.info.status == RebalStatus::Completed || !pool_stat.participating { return Ok(None); } @@ -581,17 +581,17 @@ impl ECStore { } }); - tracing::warn!("Pool {} rebalancing is started", pool_index + 1); + warn!("Pool {} rebalancing is started", pool_index + 1); while let Some(bucket) = self.next_rebal_bucket(pool_index).await? { - tracing::info!("Rebalancing bucket: {}", bucket); + warn!("Rebalancing bucket: {}", bucket); if let Err(err) = self.rebalance_bucket(rx.resubscribe(), bucket.clone(), pool_index).await { if err.to_string().contains("not initialized") { warn!("rebalance_bucket: rebalance not initialized, continue"); continue; } - tracing::error!("Error rebalancing bucket {}: {:?}", bucket, err); + error!("Error rebalancing bucket {}: {:?}", bucket, err); done_tx.send(Err(err)).await.ok(); break; } @@ -599,7 +599,7 @@ impl ECStore { self.bucket_rebalance_done(pool_index, bucket).await?; } - tracing::warn!("Pool {} rebalancing is done", pool_index + 1); + warn!("Pool {} rebalancing is done", pool_index + 1); done_tx.send(Ok(())).await.ok(); save_task.await.ok(); @@ -838,8 +838,6 @@ impl ECStore { } }; - // TODO: defer abort_multipart_upload - defer!(|| async { if let Err(err) = self .abort_multipart_upload(&bucket, &object_info.name, &res.upload_id, &ObjectOptions::default()) @@ -1068,7 +1066,7 @@ impl SetDisks { bucket: bucket.clone(), recursice: true, min_disks: listing_quorum, - agreed: Some(Box::new(move |entry: MetaCacheEntry| Box::pin(cb1(entry.clone())))), + agreed: Some(Box::new(move |entry: MetaCacheEntry| Box::pin(cb1(entry)))), partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option]| { // let cb = cb.clone(); let resolver = resolver.clone(); diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index 5a2075775..821eb5305 100644 --- a/ecstore/src/store.rs +++ b/ecstore/src/store.rs @@ -524,7 +524,9 @@ impl ECStore { // TODO: 并发 for (idx, pool) in self.pools.iter().enumerate() { - // TODO: IsSuspended + if self.is_suspended(idx).await || self.is_pool_rebalancing(idx).await { + continue; + } n_sets[idx] = pool.set_count; @@ -714,16 +716,14 @@ impl ECStore { let mut def_pool = PoolObjInfo::default(); let mut has_def_pool = false; - let pool_meta = self.pool_meta.read().await; for pinfo in ress.iter() { - if opts.skip_decommissioned && pool_meta.is_suspended(pinfo.index) { + if opts.skip_decommissioned && self.is_suspended(pinfo.index).await { continue; } - // TODO:SkipRebalancing - // if opts.SkipRebalancing && z.IsPoolRebalancing(pinfo.Index) { - // continue - // } + if opts.skip_rebalancing && self.is_pool_rebalancing(pinfo.index).await { + continue; + } if pinfo.err.is_none() { return Ok((pinfo.clone(), self.pools_with_object(&ress, opts).await)); @@ -756,15 +756,15 @@ impl ECStore { async fn pools_with_object(&self, pools: &[PoolObjInfo], opts: &ObjectOptions) -> Vec { let mut errs = Vec::new(); - let pool_meta = self.pool_meta.read().await; + for pool in pools.iter() { - if opts.skip_decommissioned && pool_meta.is_suspended(pool.index) { + if opts.skip_decommissioned && self.is_suspended(pool.index).await { + continue; + } + + if opts.skip_rebalancing && self.is_pool_rebalancing(pool.index).await { continue; } - // TODO:SkipRebalancing - // if opts.SkipRebalancing && z.IsPoolRebalancing(pinfo.Index) { - // continue - // } if let Some(err) = &pool.err { if is_err_read_quorum(err) { @@ -1865,7 +1865,9 @@ impl StorageAPI for ECStore { } for pool in self.pools.iter() { - // TODO: IsSuspended + if self.is_suspended(pool.pool_idx).await { + continue; + } let err = match pool.put_object_part(bucket, object, upload_id, part_id, data, opts).await { Ok(res) => return Ok(res), Err(err) => { @@ -1915,6 +1917,9 @@ impl StorageAPI for ECStore { let mut uploads = Vec::new(); for pool in self.pools.iter() { + if self.is_suspended(pool.pool_idx).await { + continue; + } let res = pool .list_multipart_uploads( bucket, @@ -1948,7 +1953,9 @@ impl StorageAPI for ECStore { } for (idx, pool) in self.pools.iter().enumerate() { - // // TODO: IsSuspended + if self.is_suspended(idx).await || self.is_pool_rebalancing(idx).await { + continue; + } let res = pool .list_multipart_uploads(bucket, object, None, None, None, MAX_UPLOADS_LIST) .await?; @@ -1983,8 +1990,8 @@ impl StorageAPI for ECStore { return self.pools[0].get_multipart_info(bucket, object, upload_id, opts).await; } - for (idx, pool) in self.pools.iter().enumerate() { - if self.is_suspended(idx).await { + for pool in self.pools.iter() { + if self.is_suspended(pool.pool_idx).await { continue; } @@ -2017,7 +2024,9 @@ impl StorageAPI for ECStore { } for pool in self.pools.iter() { - // TODO: IsSuspended + if self.is_suspended(pool.pool_idx).await { + continue; + } let err = match pool.abort_multipart_upload(bucket, object, upload_id, opts).await { Ok(_) => return Ok(()), @@ -2060,7 +2069,9 @@ impl StorageAPI for ECStore { } for pool in self.pools.iter() { - // TODO: IsSuspended + if self.is_suspended(pool.pool_idx).await { + continue; + } let err = match pool .complete_multipart_upload(bucket, object, upload_id, uploaded_parts.clone(), opts)