Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: close submission in session #622

Merged
merged 2 commits into from
Feb 8, 2024
Merged

feat: close submission in session #622

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
730bea6
feat: close submission in session
aneojgurhem Feb 6, 2024
1aed662
refactor: rename SubmissionClosed to SubmissionClosedException
aneojgurhem Feb 8, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 40 additions & 0 deletions Common/src/Exceptions/SubmissionClosedException.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
// This file is part of the ArmoniK project
//
// Copyright (C) ANEO, 2021-2024. All rights reserved.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY, without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.

using System;

namespace ArmoniK.Core.Common.Exceptions;

[Serializable]
public class SubmissionClosedException : ArmoniKException
{
public SubmissionClosedException()
{
}

public SubmissionClosedException(string message)
: base(message)
{
}

public SubmissionClosedException(string message,
Exception innerException)
: base(message,
innerException)
{
}
}
12 changes: 12 additions & 0 deletions Common/src/Storage/TaskLifeCycleHelper.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,18 @@ public static TaskOptions ValidateSession(SessionData sessionData,
throw exception;
}

// we are on client side
if (sessionData.SessionId == parentTaskId && !sessionData.ClientSubmission)
{
throw new SubmissionClosedException("Client submission is closed");
}

// we are on worker side
if (sessionData.SessionId != parentTaskId && !sessionData.WorkerSubmission)
{
throw new SubmissionClosedException("Worker submission is closed");
}

return localOptions;
}

Expand Down
110 changes: 60 additions & 50 deletions Common/src/gRPC/Services/GrpcTasksService.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -401,64 +401,74 @@ request.Sort is null
public override async Task<SubmitTasksResponse> SubmitTasks(SubmitTasksRequest request,
ServerCallContext context)
{
var sessionData = await sessionTable_.GetSessionAsync(request.SessionId,
context.CancellationToken)
.ConfigureAwait(false);
try
{
var sessionData = await sessionTable_.GetSessionAsync(request.SessionId,
context.CancellationToken)
.ConfigureAwait(false);


var submissionOptions = TaskLifeCycleHelper.ValidateSession(sessionData,
request.TaskOptions.ToNullableTaskOptions(),
request.SessionId,
pushQueueStorage_.MaxPriority,
logger_,
context.CancellationToken);
var submissionOptions = TaskLifeCycleHelper.ValidateSession(sessionData,
request.TaskOptions.ToNullableTaskOptions(),
request.SessionId,
pushQueueStorage_.MaxPriority,
logger_,
context.CancellationToken);

var creationRequests = request.TaskCreations.Select(creation => new TaskCreationRequest(Guid.NewGuid()
.ToString(),
creation.PayloadId,
TaskOptions.Merge(creation.TaskOptions.ToNullableTaskOptions(),
submissionOptions),
creation.ExpectedOutputKeys,
creation.DataDependencies))
.ToList();
var creationRequests = request.TaskCreations.Select(creation => new TaskCreationRequest(Guid.NewGuid()
.ToString(),
creation.PayloadId,
TaskOptions.Merge(creation.TaskOptions.ToNullableTaskOptions(),
submissionOptions),
creation.ExpectedOutputKeys,
creation.DataDependencies))
.ToList();


await TaskLifeCycleHelper.CreateTasks(taskTable_,
resultTable_,
request.SessionId,
request.SessionId,
creationRequests,
logger_,
context.CancellationToken)
.ConfigureAwait(false);
await TaskLifeCycleHelper.CreateTasks(taskTable_,
resultTable_,
request.SessionId,
request.SessionId,
creationRequests,
logger_,
context.CancellationToken)
.ConfigureAwait(false);

await TaskLifeCycleHelper.FinalizeTaskCreation(taskTable_,
resultTable_,
pushQueueStorage_,
creationRequests,
request.SessionId,
request.SessionId,
logger_,
context.CancellationToken)
.ConfigureAwait(false);
await TaskLifeCycleHelper.FinalizeTaskCreation(taskTable_,
resultTable_,
pushQueueStorage_,
creationRequests,
request.SessionId,
request.SessionId,
logger_,
context.CancellationToken)
.ConfigureAwait(false);

return new SubmitTasksResponse
{
TaskInfos =
return new SubmitTasksResponse
{
creationRequests.Select(creationRequest => new SubmitTasksResponse.Types.TaskInfo
{
DataDependencies =
{
creationRequest.DataDependencies,
},
ExpectedOutputIds =
TaskInfos =
{
creationRequests.Select(creationRequest => new SubmitTasksResponse.Types.TaskInfo
{
creationRequest.ExpectedOutputKeys,
},
TaskId = creationRequest.TaskId,
}),
},
};
DataDependencies =
{
creationRequest.DataDependencies,
},
ExpectedOutputIds =
{
creationRequest.ExpectedOutputKeys,
},
TaskId = creationRequest.TaskId,
}),
},
};
}
catch (SubmissionClosedException e)
{
logger_.LogWarning(e,
"Error while submitting tasks");
throw new RpcException(new Status(StatusCode.FailedPrecondition,
"Client submission is closed, no tasks can be submitted"));
}
}
}
103 changes: 87 additions & 16 deletions Common/tests/GrpcTasksServiceTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -229,31 +229,102 @@ await helper.App.StartAsync()
HttpHandler = server.CreateHandler(),
});

var session_id = new Sessions.SessionsClient(channel).CreateSession(new CreateSessionRequest
{
DefaultTaskOption = new TaskOptions
{
MaxRetries = 1,
Priority = 2,
MaxDuration = new Duration
{
Seconds = 500,
Nanos = 0,
},
},
})
.SessionId;
var sessionId = new Sessions.SessionsClient(channel).CreateSession(new CreateSessionRequest
{
DefaultTaskOption = new TaskOptions
{
MaxRetries = 1,
Priority = 2,
MaxDuration = new Duration
{
Seconds = 500,
Nanos = 0,
},
},
})
.SessionId;

var client = new Tasks.TasksClient(channel);
Assert.That(delegate
{
client.SubmitTasks(new SubmitTasksRequest
{
SessionId = session_id,
SessionId = sessionId,
});
},
Throws.InstanceOf<RpcException>()
.With.Property("StatusCode")
.With.Property(nameof(RpcException.StatusCode))
.EqualTo(StatusCode.InvalidArgument));
}

[Test]
public async Task SubmitTaskWithSubmissionClosedShouldFailWithRpcException()
{
var helper = new TestDatabaseProvider(collection => collection.AddSingleton<IPullQueueStorage, SimplePullQueueStorage>()
.AddSingleton<IPushQueueStorage, SimplePushQueueStorage>()
.AddSingleton<IPartitionTable, SimplePartitionTable>()
.AddSingleton<Injection.Options.Submitter>()
.AddHttpClient()
.AddGrpc(),
builder => builder.UseRouting()
.UseAuthorization(),
builder =>
{
builder.MapGrpcService<GrpcTasksService>();
builder.MapGrpcService<GrpcSessionsService>();
},
true,
true);

await helper.App.StartAsync()
.ConfigureAwait(false);

var server = helper.App.GetTestServer();

var channel = GrpcChannel.ForAddress("http://localhost:9999",
new GrpcChannelOptions
{
HttpHandler = server.CreateHandler(),
});

var sessionId = new Sessions.SessionsClient(channel).CreateSession(new CreateSessionRequest
{
DefaultTaskOption = new TaskOptions
{
MaxRetries = 1,
Priority = 2,
MaxDuration = new Duration
{
Seconds = 500,
Nanos = 0,
},
},
})
.SessionId;

new Sessions.SessionsClient(channel).StopSubmission(new StopSubmissionRequest
{
Client = true,
SessionId = sessionId,
});

var client = new Tasks.TasksClient(channel);
Assert.That(delegate
{
client.SubmitTasks(new SubmitTasksRequest
{
SessionId = sessionId,
TaskCreations =
{
new SubmitTasksRequest.Types.TaskCreation
{
PayloadId = "payload",
},
},
});
},
Throws.InstanceOf<RpcException>()
.With.Property(nameof(RpcException.StatusCode))
.EqualTo(StatusCode.FailedPrecondition));
}
}
Loading
Loading