Conversation
|
|
||
| @Override | ||
| protected TaskMonitor createTaskMonitor() { | ||
| return TaskPollingMonitor.create(session, name, 100, Duration.of("15s")) |
There was a problem hiding this comment.
del the ForchTaskMonitor? or is that going to be used later?
idk what the downsides of the TaskPollingMonitor are
|
|
||
| String command = "forch status ${forchTaskId}" | ||
| StringBuilder stdout = new StringBuilder(), stderr = new StringBuilder(); | ||
| Process proc = command.execute() |
|
|
||
| @Override | ||
| void kill() { | ||
| // noop |
There was a problem hiding this comment.
fyi you can insert a container-killed event (if it's running) or set cancelled on the task (if it's not running) to kill it
see pods-stop in nucleus
forch_interface.py
Outdated
| gpus: int | ||
|
|
||
|
|
||
| async def create_task(payload: CreateTaskInput): |
There was a problem hiding this comment.
u can just wrap this function in with_conn_retry without a helper, since it doesn't do anything else
forch_interface.py
Outdated
| ] | ||
|
|
||
|
|
||
| async def get_task_status(task_id: int): |
There was a problem hiding this comment.
wrap with @with_conn_retry directly
forch_interface.py
Outdated
| coalesce( | ||
| ( | ||
| select | ||
| ( |
There was a problem hiding this comment.
the indentation is looking sus
forch_interface.py
Outdated
| 'submitted' | ||
| when te.type = 'container-created' then | ||
| 'running' | ||
| when te.type = 'container-exited' then |
There was a problem hiding this comment.
probably also want to consider the forch_pub.tasks.cancelled_at field
forch_interface.py
Outdated
|
|
||
| async def main(): | ||
| await pool.open() | ||
| args = sys.argv[1:] |
There was a problem hiding this comment.
use argparse, it's probably easier to dela with than sys.argv
form argparse import ArgumentParser
argp = ArgumentParser("forch")
cmds = argp.add_subparsers()
argp_create = cmds.add_parser("create")
argp_create.add_argument("spec")
argp_status = cmds.add_parser("status")
argp_status.add_argument("task_id")
args = argp.parse_args()
Signed-off-by: Ayush Kamat <ayush@latch.bio>
Signed-off-by: Ayush Kamat <ayush@latch.bio>
| if( scratch==null ) | ||
| scratch = true | ||
|
|
||
| // include task script as an input to force its staging in the container work directory |
There was a problem hiding this comment.
no idea what any of this means
forch
Outdated
|
|
||
| config = read_config(AppConfig) | ||
|
|
||
| pool = get_pool(config.db, "nextflow_forch_test", read_only=False) |
forch
Outdated
| ( | ||
| select | ||
| ( | ||
| select |
There was a problem hiding this comment.
you don't need a subquery for case
forch
Outdated
| 'running' | ||
| when te.type = 'container-exited' then | ||
| ( | ||
| select |
| BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(fos)) | ||
| ) { | ||
| try (BufferedWriter writer=Files.newBufferedWriter(path, CREATE,WRITE,TRUNCATE_EXISTING)) { | ||
| writer.write(data) |
There was a problem hiding this comment.
you might still want to flush + fsync?
| local name=\$1 | ||
| local s3path=\$2 | ||
| if [[ "\$name" == - ]]; then | ||
| echo 's5cmd --no-verify-ssl pipe "\$s3path"' |
There was a problem hiding this comment.
why are we doing no-verify-ssl btw?
|
|
||
| @Override | ||
| String pipeInputFile(Path file) { | ||
| return " < ${Escape.path(file.getFileName())}" |
There was a problem hiding this comment.
| return " < ${Escape.path(file.getFileName())}" | |
| return " < ${this.fileStr(file)}" |
perhaps?
Rahuldesai1/forch gql
Signed-off-by: Ayush Kamat <ayush@latch.bio>
Signed-off-by: Ayush Kamat <ayush@latch.bio>
Signed-off-by: Ayush Kamat <ayush@latch.bio>
Signed-off-by: Ayush Kamat <ayush@latch.bio>
Signed-off-by: Ayush Kamat <ayush@latch.bio>
Signed-off-by: Ayush Kamat <ayush@latch.bio>
* ldata provenance stuff * fix: pass execution instead of token * fix * cleanup * bump latch version * throw error * debug * fix: comma * remove sleep * rename * fix
Signed-off-by: Ayush Kamat <ayush@latch.bio>
Signed-off-by: Ayush Kamat <ayush@latch.bio>
4 participants
No description provided.