#257 debugging

plantit/plantit/celery_tasks.py

@@ -185,7 +185,10 @@ def share_data(self, guid: str):
         return
 
     # if any other running tasks share the same outbound transfer path, we already have access (no need to share)
-    if Task.objects.filter(completed__isnull=True, transfer_path=task.transfer_path).count() != 0:
+    if Task.objects\
+            .filter(user=task.user, status__in=[TaskStatus.CREATED, TaskStatus.RUNNING], transfer_path=task.transfer_path)\
+            .exclude(guid=task.guid)\
+            .count() != 0:
         logger.warning(
             f"Task {guid} outbound path {task.transfer_path} used by another task, not granting temporary data access")
         self.request.callbacks = None

plantit/plantit/task_lifecycle.py

@@ -495,14 +495,14 @@ def list_result_files(task: Task) -> List[dict]:
     seen = []
     results = []
     ssh = get_task_ssh_client(task)
-    staging_dir = join(task.agent.workdir, task.workdir, f"{task.guid}_staging")
+    zip_dir = join(task.agent.workdir, task.workdir, f"{task.guid}_staging")
     expected_names = get_output_included_names(task)
     expected_patterns = get_output_included_patterns(task)
 
     with ssh:
         with ssh.client.open_sftp() as sftp:
             # list contents of task working directory
-            names = sftp.listdir(staging_dir)
+            names = sftp.listdir(zip_dir)
 
             # check for files by name
             logger.info(f"Looking for files by name: {', '.join(expected_names)}")
@@ -515,7 +515,7 @@ def list_result_files(task: Task) -> List[dict]:
 
                 output = {
                     'name': name,
-                    'path': join(staging_dir, name),
+                    'path': join(zip_dir, name),
                     'exists': exists
                 }
                 results.append(output)
@@ -542,7 +542,7 @@ def list_result_files(task: Task) -> List[dict]:
                         if not any(s == name for s in seen):
                             results.append({
                                 'name': name,
-                                'path': join(staging_dir, name),
+                                'path': join(zip_dir, name),
                                 'exists': True
                             })
 
@@ -553,7 +553,7 @@ def list_result_files(task: Task) -> List[dict]:
                 if not any_matched:
                     results.append({
                         'name': f"*.{pattern}",
-                        'path': join(staging_dir, f"*.{pattern}"),
+                        'path': join(zip_dir, f"*.{pattern}"),
                         'exists': False
                     })
 

plantit/plantit/task_scripts.py

@@ -354,18 +354,15 @@ def compose_push_commands(task: Task, options: TaskOptions) -> List[str]:
     zip_command = f"zip -r {zip_path} {zip_dir}/*"
     commands.append(zip_command)
 
-    # move zip file into staging dir
-    mv_zip_command = f"mv {zip_path} {join(staging_dir, zip_name)}"
-    commands.append(mv_zip_command)
-
     # transfer contents of staging dir to CyVerse
-    path = output['to']
+    to_path = output['to']
     image = f"docker://{settings.ICOMMANDS_IMAGE}"
     # force = output['force']
     force = False
     # just_zip = output['just_zip']
     just_zip = False
-    push_command = f"singularity exec {image} iput{' -f ' if force else ' '}{staging_dir}{('/' + zip_name) if just_zip else '/*'} {path}"
+    # push_command = f"singularity exec {image} iput -r{' -f ' if force else ' '}{staging_dir}{('/' + zip_name) if just_zip else '/*'} {to_path}/"
+    push_command = f"singularity exec {image} iput -f {staging_dir}{('/' + zip_name) if just_zip else '/*'} {to_path}/"
     commands.append(push_command)
 
     newline = '\n'