feat: add ReadPersistentCacheTask and WritePersistentCacheTask message (#441)

Signed-off-by: Gaius <gaius.qi@gmail.com>

Cargo.lock

@@ -190,7 +190,7 @@ dependencies = [
 
 [[package]]
 name = "dragonfly-api"
-version = "2.1.4"
+version = "2.1.5"
 dependencies = [
  "prost",
  "prost-types",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "dragonfly-api"
-version = "2.1.4"
+version = "2.1.5"
 authors = ["Gaius <gaius.qi@gmail.com>"]
 edition = "2021"
 license = "Apache-2.0"

pkg/apis/dfdaemon/v2/dfdaemon.proto

@@ -125,6 +125,71 @@ message DeleteTaskRequest {
   string task_id = 1 [(validate.rules).string.min_len = 1];
 }
 
+// WritePersistentCacheTaskResponse represents request of WritePersistentCacheTask.
+message WritePersistentCacheTaskRequest {
+  oneof response {
+    option (validate.required) = true;
+
+    WritePersistentCacheTaskStartedRequest write_persistent_cache_task_started_request = 1;
+    WritePersistentCacheTaskFinishedRequest write_persistent_cache_task_finished_request = 2;
+    WriteChunkRequest write_chunk_request = 3;
+  }
+}
+
+// WritePersistentCacheTaskStartedRequest represents task write started request of WritePersistentCacheTaskRequest.
+message WritePersistentCacheTaskStartedRequest {
+  // Replica count of the persistent persistent cache task.
+  uint64 persistent_replica_count = 1 [(validate.rules).uint64 = {gte: 1, lte: 5}];
+  // Tag is used to distinguish different persistent cache tasks.
+  optional string tag = 2;
+  // Application of task.
+  optional string application = 3;
+  // TTL of the persistent cache task.
+  google.protobuf.Duration ttl = 4 [(validate.rules).duration = {gte:{seconds: 60}, lte:{seconds: 604800}}];
+}
+
+// WritePersistentCacheTaskFinishedRequest represents task write finished request of WritePersistentCacheTaskRequest.
+message WritePersistentCacheTaskFinishedRequest {
+}
+
+// WriteChunkRequest represents chunk write request of WritePersistentCacheTaskRequest.
+message WriteChunkRequest {
+  // Chunk content.
+  bytes content = 1 [(validate.rules).bytes = {min_len: 1}];
+}
+
+// WritePersistentCacheTaskResponse represents response of WritePersistentCacheTask.
+message WritePersistentCacheTaskResponse {
+  // Task id.
+  string task_id = 1 [(validate.rules).string.min_len = 1];
+}
+
+// ReadPersistentCacheTaskRequest represents request of ReadPersistentCacheTask.
+message ReadPersistentCacheTaskRequest {
+  // Task id.
+  string task_id = 1 [(validate.rules).string.min_len = 1];
+}
+
+// ReadPersistentCacheTaskResponse represents response of ReadPersistentCacheTask.
+message ReadPersistentCacheTaskResponse {
+  oneof response {
+    option (validate.required) = true;
+
+    ReadPersistentCacheTaskFinishedResponse read_persistent_cache_task_finished_response = 1;
+    ReadChunkResponse read_chunk_response = 2;
+  }
+}
+
+// ReadPersistentCacheTaskFinishedResponse represents task read finished response of ReadPersistentCacheTaskResponse.
+message ReadPersistentCacheTaskFinishedResponse {
+}
+
+// ReadChunkResponse represents chunk read response of ReadPersistentCacheTaskResponse.
+message ReadChunkResponse {
+  // Chunk content.
+  bytes content = 1 [(validate.rules).bytes = {min_len: 1}];
+}
+
 // DownloadPersistentCacheTaskRequest represents request of DownloadPersistentCacheTask.
 message DownloadPersistentCacheTaskRequest {
   // Task id.
@@ -173,15 +238,13 @@ message UploadPersistentCacheTaskRequest {
   // Upload file path of persistent cache task.
   string path = 1 [(validate.rules).string = {min_len: 1}];
   // Replica count of the persistent persistent cache task.
-  uint64 persistent_replica_count = 2 [(validate.rules).uint64.gte = 1];
+  uint64 persistent_replica_count = 2 [(validate.rules).uint64 = {gte: 1, lte: 5}];
   // Tag is used to distinguish different persistent cache tasks.
   optional string tag = 3;
   // Application of task.
   optional string application = 4;
   // TTL of the persistent cache task.
   google.protobuf.Duration ttl = 5 [(validate.rules).duration = {gte:{seconds: 60}, lte:{seconds: 604800}}];
-  // Upload timeout.
-  optional google.protobuf.Duration timeout = 6;
 }
 
 // StatPersistentCacheTaskRequest represents request of StatPersistentCacheTask.
@@ -251,6 +314,12 @@ service DfdaemonDownload {
   // DeleteHost releases host in scheduler.
   rpc DeleteHost(google.protobuf.Empty)returns(google.protobuf.Empty);
 
+  // WritePersistentCacheTask writes persistent cache task to p2p network.
+  rpc WritePersistentCacheTask(stream WritePersistentCacheTaskRequest) returns(WritePersistentCacheTaskResponse);
+
+  // ReadPersistentCacheTask reads persistent cache task from p2p network.
+  rpc ReadPersistentCacheTask(ReadPersistentCacheTaskRequest) returns(stream ReadPersistentCacheTaskResponse);
+
   // DownloadPersistentCacheTask downloads persistent cache task from p2p network.
   rpc DownloadPersistentCacheTask(DownloadPersistentCacheTaskRequest) returns(stream DownloadPersistentCacheTaskResponse);
 

pkg/apis/dfdaemon/v2/dfdaemon_grpc.pb.go

@@ -533,6 +533,10 @@ type DfdaemonDownloadClient interface {
 	DeleteTask(ctx context.Context, in *DeleteTaskRequest, opts ...grpc.CallOption) (*emptypb.Empty, error)
 	// DeleteHost releases host in scheduler.
 	DeleteHost(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*emptypb.Empty, error)
+	// WritePersistentCacheTask writes persistent cache task to p2p network.
+	WritePersistentCacheTask(ctx context.Context, opts ...grpc.CallOption) (DfdaemonDownload_WritePersistentCacheTaskClient, error)
+	// ReadPersistentCacheTask reads persistent cache task from p2p network.
+	ReadPersistentCacheTask(ctx context.Context, in *ReadPersistentCacheTaskRequest, opts ...grpc.CallOption) (DfdaemonDownload_ReadPersistentCacheTaskClient, error)
 	// DownloadPersistentCacheTask downloads persistent cache task from p2p network.
 	DownloadPersistentCacheTask(ctx context.Context, in *DownloadPersistentCacheTaskRequest, opts ...grpc.CallOption) (DfdaemonDownload_DownloadPersistentCacheTaskClient, error)
 	// UploadPersistentCacheTask uploads persistent cache task to p2p network.
@@ -619,8 +623,74 @@ func (c *dfdaemonDownloadClient) DeleteHost(ctx context.Context, in *emptypb.Emp
 	return out, nil
 }
 
+func (c *dfdaemonDownloadClient) WritePersistentCacheTask(ctx context.Context, opts ...grpc.CallOption) (DfdaemonDownload_WritePersistentCacheTaskClient, error) {
+	stream, err := c.cc.NewStream(ctx, &DfdaemonDownload_ServiceDesc.Streams[1], "/dfdaemon.v2.DfdaemonDownload/WritePersistentCacheTask", opts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &dfdaemonDownloadWritePersistentCacheTaskClient{stream}
+	return x, nil
+}
+
+type DfdaemonDownload_WritePersistentCacheTaskClient interface {
+	Send(*WritePersistentCacheTaskRequest) error
+	CloseAndRecv() (*WritePersistentCacheTaskResponse, error)
+	grpc.ClientStream
+}
+
+type dfdaemonDownloadWritePersistentCacheTaskClient struct {
+	grpc.ClientStream
+}
+
+func (x *dfdaemonDownloadWritePersistentCacheTaskClient) Send(m *WritePersistentCacheTaskRequest) error {
+	return x.ClientStream.SendMsg(m)
+}
+
+func (x *dfdaemonDownloadWritePersistentCacheTaskClient) CloseAndRecv() (*WritePersistentCacheTaskResponse, error) {
+	if err := x.ClientStream.CloseSend(); err != nil {
+		return nil, err
+	}
+	m := new(WritePersistentCacheTaskResponse)
+	if err := x.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+func (c *dfdaemonDownloadClient) ReadPersistentCacheTask(ctx context.Context, in *ReadPersistentCacheTaskRequest, opts ...grpc.CallOption) (DfdaemonDownload_ReadPersistentCacheTaskClient, error) {
+	stream, err := c.cc.NewStream(ctx, &DfdaemonDownload_ServiceDesc.Streams[2], "/dfdaemon.v2.DfdaemonDownload/ReadPersistentCacheTask", opts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &dfdaemonDownloadReadPersistentCacheTaskClient{stream}
+	if err := x.ClientStream.SendMsg(in); err != nil {
+		return nil, err
+	}
+	if err := x.ClientStream.CloseSend(); err != nil {
+		return nil, err
+	}
+	return x, nil
+}
+
+type DfdaemonDownload_ReadPersistentCacheTaskClient interface {
+	Recv() (*ReadPersistentCacheTaskResponse, error)
+	grpc.ClientStream
+}
+
+type dfdaemonDownloadReadPersistentCacheTaskClient struct {
+	grpc.ClientStream
+}
+
+func (x *dfdaemonDownloadReadPersistentCacheTaskClient) Recv() (*ReadPersistentCacheTaskResponse, error) {
+	m := new(ReadPersistentCacheTaskResponse)
+	if err := x.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
 func (c *dfdaemonDownloadClient) DownloadPersistentCacheTask(ctx context.Context, in *DownloadPersistentCacheTaskRequest, opts ...grpc.CallOption) (DfdaemonDownload_DownloadPersistentCacheTaskClient, error) {
-	stream, err := c.cc.NewStream(ctx, &DfdaemonDownload_ServiceDesc.Streams[1], "/dfdaemon.v2.DfdaemonDownload/DownloadPersistentCacheTask", opts...)
+	stream, err := c.cc.NewStream(ctx, &DfdaemonDownload_ServiceDesc.Streams[3], "/dfdaemon.v2.DfdaemonDownload/DownloadPersistentCacheTask", opts...)
 	if err != nil {
 		return nil, err
 	}
@@ -692,6 +762,10 @@ type DfdaemonDownloadServer interface {
 	DeleteTask(context.Context, *DeleteTaskRequest) (*emptypb.Empty, error)
 	// DeleteHost releases host in scheduler.
 	DeleteHost(context.Context, *emptypb.Empty) (*emptypb.Empty, error)
+	// WritePersistentCacheTask writes persistent cache task to p2p network.
+	WritePersistentCacheTask(DfdaemonDownload_WritePersistentCacheTaskServer) error
+	// ReadPersistentCacheTask reads persistent cache task from p2p network.
+	ReadPersistentCacheTask(*ReadPersistentCacheTaskRequest, DfdaemonDownload_ReadPersistentCacheTaskServer) error
 	// DownloadPersistentCacheTask downloads persistent cache task from p2p network.
 	DownloadPersistentCacheTask(*DownloadPersistentCacheTaskRequest, DfdaemonDownload_DownloadPersistentCacheTaskServer) error
 	// UploadPersistentCacheTask uploads persistent cache task to p2p network.
@@ -721,6 +795,12 @@ func (UnimplementedDfdaemonDownloadServer) DeleteTask(context.Context, *DeleteTa
 func (UnimplementedDfdaemonDownloadServer) DeleteHost(context.Context, *emptypb.Empty) (*emptypb.Empty, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method DeleteHost not implemented")
 }
+func (UnimplementedDfdaemonDownloadServer) WritePersistentCacheTask(DfdaemonDownload_WritePersistentCacheTaskServer) error {
+	return status.Errorf(codes.Unimplemented, "method WritePersistentCacheTask not implemented")
+}
+func (UnimplementedDfdaemonDownloadServer) ReadPersistentCacheTask(*ReadPersistentCacheTaskRequest, DfdaemonDownload_ReadPersistentCacheTaskServer) error {
+	return status.Errorf(codes.Unimplemented, "method ReadPersistentCacheTask not implemented")
+}
 func (UnimplementedDfdaemonDownloadServer) DownloadPersistentCacheTask(*DownloadPersistentCacheTaskRequest, DfdaemonDownload_DownloadPersistentCacheTaskServer) error {
 	return status.Errorf(codes.Unimplemented, "method DownloadPersistentCacheTask not implemented")
 }
@@ -838,6 +918,53 @@ func _DfdaemonDownload_DeleteHost_Handler(srv interface{}, ctx context.Context,
 	return interceptor(ctx, in, info, handler)
 }
 
+func _DfdaemonDownload_WritePersistentCacheTask_Handler(srv interface{}, stream grpc.ServerStream) error {
+	return srv.(DfdaemonDownloadServer).WritePersistentCacheTask(&dfdaemonDownloadWritePersistentCacheTaskServer{stream})
+}
+
+type DfdaemonDownload_WritePersistentCacheTaskServer interface {
+	SendAndClose(*WritePersistentCacheTaskResponse) error
+	Recv() (*WritePersistentCacheTaskRequest, error)
+	grpc.ServerStream
+}
+
+type dfdaemonDownloadWritePersistentCacheTaskServer struct {
+	grpc.ServerStream
+}
+
+func (x *dfdaemonDownloadWritePersistentCacheTaskServer) SendAndClose(m *WritePersistentCacheTaskResponse) error {
+	return x.ServerStream.SendMsg(m)
+}
+
+func (x *dfdaemonDownloadWritePersistentCacheTaskServer) Recv() (*WritePersistentCacheTaskRequest, error) {
+	m := new(WritePersistentCacheTaskRequest)
+	if err := x.ServerStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+func _DfdaemonDownload_ReadPersistentCacheTask_Handler(srv interface{}, stream grpc.ServerStream) error {
+	m := new(ReadPersistentCacheTaskRequest)
+	if err := stream.RecvMsg(m); err != nil {
+		return err
+	}
+	return srv.(DfdaemonDownloadServer).ReadPersistentCacheTask(m, &dfdaemonDownloadReadPersistentCacheTaskServer{stream})
+}
+
+type DfdaemonDownload_ReadPersistentCacheTaskServer interface {
+	Send(*ReadPersistentCacheTaskResponse) error
+	grpc.ServerStream
+}
+
+type dfdaemonDownloadReadPersistentCacheTaskServer struct {
+	grpc.ServerStream
+}
+
+func (x *dfdaemonDownloadReadPersistentCacheTaskServer) Send(m *ReadPersistentCacheTaskResponse) error {
+	return x.ServerStream.SendMsg(m)
+}
+
 func _DfdaemonDownload_DownloadPersistentCacheTask_Handler(srv interface{}, stream grpc.ServerStream) error {
 	m := new(DownloadPersistentCacheTaskRequest)
 	if err := stream.RecvMsg(m); err != nil {
@@ -955,6 +1082,16 @@ var DfdaemonDownload_ServiceDesc = grpc.ServiceDesc{
 			Handler:       _DfdaemonDownload_DownloadTask_Handler,
 			ServerStreams: true,
 		},
+		{
+			StreamName:    "WritePersistentCacheTask",
+			Handler:       _DfdaemonDownload_WritePersistentCacheTask_Handler,
+			ClientStreams: true,
+		},
+		{
+			StreamName:    "ReadPersistentCacheTask",
+			Handler:       _DfdaemonDownload_ReadPersistentCacheTask_Handler,
+			ServerStreams: true,
+		},
 		{
 			StreamName:    "DownloadPersistentCacheTask",
 			Handler:       _DfdaemonDownload_DownloadPersistentCacheTask_Handler,

proto/dfdaemon.proto

@@ -114,6 +114,67 @@ message DeleteTaskRequest {
   string task_id = 1;
 }
 
+// WritePersistentCacheTaskResponse represents request of WritePersistentCacheTask.
+message WritePersistentCacheTaskRequest {
+  oneof response {
+    WritePersistentCacheTaskStartedRequest write_persistent_cache_task_started_request = 1;
+    WritePersistentCacheTaskFinishedRequest write_persistent_cache_task_finished_request = 2;
+    WriteChunkRequest write_chunk_request = 3;
+  }
+}
+
+// WritePersistentCacheTaskStartedRequest represents task write started request of WritePersistentCacheTaskRequest.
+message WritePersistentCacheTaskStartedRequest {
+  // Replica count of the persistent persistent cache task.
+  uint64 persistent_replica_count = 1;
+  // Tag is used to distinguish different persistent cache tasks.
+  optional string tag = 2;
+  // Application of task.
+  optional string application = 3;
+  // TTL of the persistent cache task.
+  google.protobuf.Duration ttl = 4;
+}
+
+// WritePersistentCacheTaskFinishedRequest represents task write finished request of WritePersistentCacheTaskRequest.
+message WritePersistentCacheTaskFinishedRequest {
+}
+
+// WriteChunkRequest represents chunk write request of WritePersistentCacheTaskRequest.
+message WriteChunkRequest {
+  // Chunk content.
+  bytes content = 1;
+}
+
+// WritePersistentCacheTaskResponse represents response of WritePersistentCacheTask.
+message WritePersistentCacheTaskResponse {
+  // Task id.
+  string task_id = 1;
+}
+
+// ReadPersistentCacheTaskRequest represents request of ReadPersistentCacheTask.
+message ReadPersistentCacheTaskRequest {
+  // Task id.
+  string task_id = 1;
+}
+
+// ReadPersistentCacheTaskResponse represents response of ReadPersistentCacheTask.
+message ReadPersistentCacheTaskResponse {
+  oneof response {
+    ReadPersistentCacheTaskFinishedResponse read_persistent_cache_task_finished_response = 1;
+    ReadChunkResponse read_chunk_response = 2;
+  }
+}
+
+// ReadPersistentCacheTaskFinishedResponse represents task read finished response of ReadPersistentCacheTaskResponse.
+message ReadPersistentCacheTaskFinishedResponse {
+}
+
+// ReadChunkResponse represents chunk read response of ReadPersistentCacheTaskResponse.
+message ReadChunkResponse {
+  // Chunk content.
+  bytes content = 1;
+}
+
 // DownloadPersistentCacheTaskRequest represents request of DownloadPersistentCacheTask.
 message DownloadPersistentCacheTaskRequest {
   // Task id.
@@ -167,8 +228,6 @@ message UploadPersistentCacheTaskRequest {
   optional string application = 4;
   // TTL of the persistent cache task.
   google.protobuf.Duration ttl = 5;
-  // Upload timeout.
-  optional google.protobuf.Duration timeout = 6;
 }
 
 // StatPersistentCacheTaskRequest represents request of StatPersistentCacheTask.
@@ -235,6 +294,12 @@ service DfdaemonDownload{
   // DeleteHost releases host in scheduler.
   rpc DeleteHost(google.protobuf.Empty)returns(google.protobuf.Empty);
 
+  // WritePersistentCacheTask writes persistent cache task to p2p network.
+  rpc WritePersistentCacheTask(stream WritePersistentCacheTaskRequest) returns(WritePersistentCacheTaskResponse);
+
+  // ReadPersistentCacheTask reads persistent cache task from p2p network.
+  rpc ReadPersistentCacheTask(ReadPersistentCacheTaskRequest) returns(stream ReadPersistentCacheTaskResponse);
+
   // DownloadPersistentCacheTask downloads persistent cache task from p2p network.
   rpc DownloadPersistentCacheTask(DownloadPersistentCacheTaskRequest) returns(stream DownloadPersistentCacheTaskResponse);
 

src/dfdaemon.v2.rs

@@ -141,6 +141,118 @@ pub struct DeleteTaskRequest {
     #[prost(string, tag = "1")]
     pub task_id: ::prost::alloc::string::String,
 }
+/// WritePersistentCacheTaskResponse represents request of WritePersistentCacheTask.
+#[derive(serde::Serialize, serde::Deserialize)]
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct WritePersistentCacheTaskRequest {
+    #[prost(oneof = "write_persistent_cache_task_request::Response", tags = "1, 2, 3")]
+    pub response: ::core::option::Option<write_persistent_cache_task_request::Response>,
+}
+/// Nested message and enum types in `WritePersistentCacheTaskRequest`.
+pub mod write_persistent_cache_task_request {
+    #[derive(serde::Serialize, serde::Deserialize)]
+    #[allow(clippy::derive_partial_eq_without_eq)]
+    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    pub enum Response {
+        #[prost(message, tag = "1")]
+        WritePersistentCacheTaskStartedRequest(
+            super::WritePersistentCacheTaskStartedRequest,
+        ),
+        #[prost(message, tag = "2")]
+        WritePersistentCacheTaskFinishedRequest(
+            super::WritePersistentCacheTaskFinishedRequest,
+        ),
+        #[prost(message, tag = "3")]
+        WriteChunkRequest(super::WriteChunkRequest),
+    }
+}
+/// WritePersistentCacheTaskStartedRequest represents task write started request of WritePersistentCacheTaskRequest.
+#[derive(serde::Serialize, serde::Deserialize)]
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct WritePersistentCacheTaskStartedRequest {
+    /// Replica count of the persistent persistent cache task.
+    #[prost(uint64, tag = "1")]
+    pub persistent_replica_count: u64,
+    /// Tag is used to distinguish different persistent cache tasks.
+    #[prost(string, optional, tag = "2")]
+    pub tag: ::core::option::Option<::prost::alloc::string::String>,
+    /// Application of task.
+    #[prost(string, optional, tag = "3")]
+    pub application: ::core::option::Option<::prost::alloc::string::String>,
+    /// TTL of the persistent cache task.
+    #[prost(message, optional, tag = "4")]
+    pub ttl: ::core::option::Option<::prost_wkt_types::Duration>,
+}
+/// WritePersistentCacheTaskFinishedRequest represents task write finished request of WritePersistentCacheTaskRequest.
+#[derive(serde::Serialize, serde::Deserialize)]
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
+pub struct WritePersistentCacheTaskFinishedRequest {}
+/// WriteChunkRequest represents chunk write request of WritePersistentCacheTaskRequest.
+#[derive(serde::Serialize, serde::Deserialize)]
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct WriteChunkRequest {
+    /// Chunk content.
+    #[prost(bytes = "vec", tag = "1")]
+    pub content: ::prost::alloc::vec::Vec<u8>,
+}
+/// WritePersistentCacheTaskResponse represents response of WritePersistentCacheTask.
+#[derive(serde::Serialize, serde::Deserialize)]
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct WritePersistentCacheTaskResponse {
+    /// Task id.
+    #[prost(string, tag = "1")]
+    pub task_id: ::prost::alloc::string::String,
+}
+/// ReadPersistentCacheTaskRequest represents request of ReadPersistentCacheTask.
+#[derive(serde::Serialize, serde::Deserialize)]
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ReadPersistentCacheTaskRequest {
+    /// Task id.
+    #[prost(string, tag = "1")]
+    pub task_id: ::prost::alloc::string::String,
+}
+/// ReadPersistentCacheTaskResponse represents response of ReadPersistentCacheTask.
+#[derive(serde::Serialize, serde::Deserialize)]
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ReadPersistentCacheTaskResponse {
+    #[prost(oneof = "read_persistent_cache_task_response::Response", tags = "1, 2")]
+    pub response: ::core::option::Option<read_persistent_cache_task_response::Response>,
+}
+/// Nested message and enum types in `ReadPersistentCacheTaskResponse`.
+pub mod read_persistent_cache_task_response {
+    #[derive(serde::Serialize, serde::Deserialize)]
+    #[allow(clippy::derive_partial_eq_without_eq)]
+    #[derive(Clone, PartialEq, ::prost::Oneof)]
+    pub enum Response {
+        #[prost(message, tag = "1")]
+        ReadPersistentCacheTaskFinishedResponse(
+            super::ReadPersistentCacheTaskFinishedResponse,
+        ),
+        #[prost(message, tag = "2")]
+        ReadChunkResponse(super::ReadChunkResponse),
+    }
+}
+/// ReadPersistentCacheTaskFinishedResponse represents task read finished response of ReadPersistentCacheTaskResponse.
+#[derive(serde::Serialize, serde::Deserialize)]
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, Copy, PartialEq, ::prost::Message)]
+pub struct ReadPersistentCacheTaskFinishedResponse {}
+/// ReadChunkResponse represents chunk read response of ReadPersistentCacheTaskResponse.
+#[derive(serde::Serialize, serde::Deserialize)]
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Clone, PartialEq, ::prost::Message)]
+pub struct ReadChunkResponse {
+    /// Chunk content.
+    #[prost(bytes = "vec", tag = "1")]
+    pub content: ::prost::alloc::vec::Vec<u8>,
+}
 /// DownloadPersistentCacheTaskRequest represents request of DownloadPersistentCacheTask.
 #[derive(serde::Serialize, serde::Deserialize)]
 #[allow(clippy::derive_partial_eq_without_eq)]
@@ -232,9 +344,6 @@ pub struct UploadPersistentCacheTaskRequest {
     /// TTL of the persistent cache task.
     #[prost(message, optional, tag = "5")]
     pub ttl: ::core::option::Option<::prost_wkt_types::Duration>,
-    /// Upload timeout.
-    #[prost(message, optional, tag = "6")]
-    pub timeout: ::core::option::Option<::prost_wkt_types::Duration>,
 }
 /// StatPersistentCacheTaskRequest represents request of StatPersistentCacheTask.
 #[derive(serde::Serialize, serde::Deserialize)]
@@ -785,6 +894,72 @@ pub mod dfdaemon_download_client {
                 .insert(GrpcMethod::new("dfdaemon.v2.DfdaemonDownload", "DeleteHost"));
             self.inner.unary(req, path, codec).await
         }
+        /// WritePersistentCacheTask writes persistent cache task to p2p network.
+        pub async fn write_persistent_cache_task(
+            &mut self,
+            request: impl tonic::IntoStreamingRequest<
+                Message = super::WritePersistentCacheTaskRequest,
+            >,
+        ) -> std::result::Result<
+            tonic::Response<super::WritePersistentCacheTaskResponse>,
+            tonic::Status,
+        > {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| {
+                    tonic::Status::new(
+                        tonic::Code::Unknown,
+                        format!("Service was not ready: {}", e.into()),
+                    )
+                })?;
+            let codec = tonic::codec::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static(
+                "/dfdaemon.v2.DfdaemonDownload/WritePersistentCacheTask",
+            );
+            let mut req = request.into_streaming_request();
+            req.extensions_mut()
+                .insert(
+                    GrpcMethod::new(
+                        "dfdaemon.v2.DfdaemonDownload",
+                        "WritePersistentCacheTask",
+                    ),
+                );
+            self.inner.client_streaming(req, path, codec).await
+        }
+        /// ReadPersistentCacheTask reads persistent cache task from p2p network.
+        pub async fn read_persistent_cache_task(
+            &mut self,
+            request: impl tonic::IntoRequest<super::ReadPersistentCacheTaskRequest>,
+        ) -> std::result::Result<
+            tonic::Response<
+                tonic::codec::Streaming<super::ReadPersistentCacheTaskResponse>,
+            >,
+            tonic::Status,
+        > {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| {
+                    tonic::Status::new(
+                        tonic::Code::Unknown,
+                        format!("Service was not ready: {}", e.into()),
+                    )
+                })?;
+            let codec = tonic::codec::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static(
+                "/dfdaemon.v2.DfdaemonDownload/ReadPersistentCacheTask",
+            );
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(
+                    GrpcMethod::new(
+                        "dfdaemon.v2.DfdaemonDownload",
+                        "ReadPersistentCacheTask",
+                    ),
+                );
+            self.inner.server_streaming(req, path, codec).await
+        }
         /// DownloadPersistentCacheTask downloads persistent cache task from p2p network.
         pub async fn download_persistent_cache_task(
             &mut self,
@@ -1589,6 +1764,33 @@ pub mod dfdaemon_download_server {
             &self,
             request: tonic::Request<()>,
         ) -> std::result::Result<tonic::Response<()>, tonic::Status>;
+        /// WritePersistentCacheTask writes persistent cache task to p2p network.
+        async fn write_persistent_cache_task(
+            &self,
+            request: tonic::Request<
+                tonic::Streaming<super::WritePersistentCacheTaskRequest>,
+            >,
+        ) -> std::result::Result<
+            tonic::Response<super::WritePersistentCacheTaskResponse>,
+            tonic::Status,
+        >;
+        /// Server streaming response type for the ReadPersistentCacheTask method.
+        type ReadPersistentCacheTaskStream: tonic::codegen::tokio_stream::Stream<
+                Item = std::result::Result<
+                    super::ReadPersistentCacheTaskResponse,
+                    tonic::Status,
+                >,
+            >
+            + std::marker::Send
+            + 'static;
+        /// ReadPersistentCacheTask reads persistent cache task from p2p network.
+        async fn read_persistent_cache_task(
+            &self,
+            request: tonic::Request<super::ReadPersistentCacheTaskRequest>,
+        ) -> std::result::Result<
+            tonic::Response<Self::ReadPersistentCacheTaskStream>,
+            tonic::Status,
+        >;
         /// Server streaming response type for the DownloadPersistentCacheTask method.
         type DownloadPersistentCacheTaskStream: tonic::codegen::tokio_stream::Stream<
                 Item = std::result::Result<
@@ -1882,6 +2084,111 @@ pub mod dfdaemon_download_server {
                     };
                     Box::pin(fut)
                 }
+                "/dfdaemon.v2.DfdaemonDownload/WritePersistentCacheTask" => {
+                    #[allow(non_camel_case_types)]
+                    struct WritePersistentCacheTaskSvc<T: DfdaemonDownload>(pub Arc<T>);
+                    impl<
+                        T: DfdaemonDownload,
+                    > tonic::server::ClientStreamingService<
+                        super::WritePersistentCacheTaskRequest,
+                    > for WritePersistentCacheTaskSvc<T> {
+                        type Response = super::WritePersistentCacheTaskResponse;
+                        type Future = BoxFuture<
+                            tonic::Response<Self::Response>,
+                            tonic::Status,
+                        >;
+                        fn call(
+                            &mut self,
+                            request: tonic::Request<
+                                tonic::Streaming<super::WritePersistentCacheTaskRequest>,
+                            >,
+                        ) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move {
+                                <T as DfdaemonDownload>::write_persistent_cache_task(
+                                        &inner,
+                                        request,
+                                    )
+                                    .await
+                            };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = WritePersistentCacheTaskSvc(inner);
+                        let codec = tonic::codec::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(
+                                accept_compression_encodings,
+                                send_compression_encodings,
+                            )
+                            .apply_max_message_size_config(
+                                max_decoding_message_size,
+                                max_encoding_message_size,
+                            );
+                        let res = grpc.client_streaming(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
+                "/dfdaemon.v2.DfdaemonDownload/ReadPersistentCacheTask" => {
+                    #[allow(non_camel_case_types)]
+                    struct ReadPersistentCacheTaskSvc<T: DfdaemonDownload>(pub Arc<T>);
+                    impl<
+                        T: DfdaemonDownload,
+                    > tonic::server::ServerStreamingService<
+                        super::ReadPersistentCacheTaskRequest,
+                    > for ReadPersistentCacheTaskSvc<T> {
+                        type Response = super::ReadPersistentCacheTaskResponse;
+                        type ResponseStream = T::ReadPersistentCacheTaskStream;
+                        type Future = BoxFuture<
+                            tonic::Response<Self::ResponseStream>,
+                            tonic::Status,
+                        >;
+                        fn call(
+                            &mut self,
+                            request: tonic::Request<
+                                super::ReadPersistentCacheTaskRequest,
+                            >,
+                        ) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move {
+                                <T as DfdaemonDownload>::read_persistent_cache_task(
+                                        &inner,
+                                        request,
+                                    )
+                                    .await
+                            };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = ReadPersistentCacheTaskSvc(inner);
+                        let codec = tonic::codec::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(
+                                accept_compression_encodings,
+                                send_compression_encodings,
+                            )
+                            .apply_max_message_size_config(
+                                max_decoding_message_size,
+                                max_encoding_message_size,
+                            );
+                        let res = grpc.server_streaming(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
                 "/dfdaemon.v2.DfdaemonDownload/DownloadPersistentCacheTask" => {
                     #[allow(non_camel_case_types)]
                     struct DownloadPersistentCacheTaskSvc<T: DfdaemonDownload>(