OpenSwathWorkflowScheduler.h

Go to the documentation of this file.

// Copyright (c) 2002-present, OpenMS Inc. -- EKU Tuebingen, ETH Zurich, and FU Berlin
// SPDX-License-Identifier: BSD-3-Clause
//
// --------------------------------------------------------------------------
// $Maintainer: Justin Sing $
// $Authors: Justin Sing $
// --------------------------------------------------------------------------
 
#pragma once
 
#include <OpenMS/CONCEPT/Types.h>
#include <OpenMS/OPENSWATHALGO/DATAACCESS/SwathMap.h>
 
#include <condition_variable>
#include <mutex>
#include <vector>
 
namespace OpenMS
{
  class OPENMS_DLLAPI OpenSwathWorkflowScheduler
  {
  public:
    struct OPENMS_DLLAPI Options
    {
      UInt64 osw_buffer_bytes = 2ULL * 1024ULL * 1024ULL * 1024ULL;
      double memory_usage_fraction = 0.90;
      UInt64 bytes_per_spectrum = 600ULL * 1024ULL;
      UInt64 per_swath_overhead_bytes = 100ULL * 1024ULL * 1024ULL;
      Size avg_transitions_per_swath = 0;
      UInt64 bytes_per_chromatogram_point = 64ULL;
      int max_concurrent_swaths = -1;
      Size scoring_threads = 1;
      Size min_inner_batch_size = 2000;
      Size max_inner_batch_size = 10000;
      Size target_jobs_per_thread = 3;
    };
 
    struct OPENMS_DLLAPI ConcurrencyEstimate
    {
      Size non_ms1_swath_count = 0;        
      Size avg_spectra_per_swath = 0;      
      Size max_concurrent_swaths = 1;      
      UInt64 memory_budget_bytes = 0;      
      UInt64 estimated_bytes_per_swath = 0;
    };
 
    struct OPENMS_DLLAPI Wave
    {
      std::vector<Size> swath_indices; 
      UInt64 estimated_bytes = 0;      
    };
 
    static UInt64 estimateAvailableMemoryForScoring(const Options& options);
 
    static ConcurrencyEstimate estimateConcurrency(
      const std::vector<OpenSwath::SwathMap>& swath_maps,
      const Options& options);
 
    static std::vector<Wave> planWaves(
      const std::vector<OpenSwath::SwathMap>& swath_maps,
      const Options& options,
      const ConcurrencyEstimate& estimate);
 
    static Size chooseInnerBatchSize(
      Size total_compounds,
      Size active_swaths,
      Size scoring_threads,
      int user_inner_batch_size,
      const Options& options);
 
    class OPENMS_DLLAPI ConcurrencyLimiter
    {
    public:
      explicit ConcurrencyLimiter(Size max_concurrent_swaths);
      void acquireSlot();
      void releaseSlot();
 
    private:
      Size max_concurrent_swaths_;
      Size active_swaths_ = 0;
      std::mutex mutex_;
      std::condition_variable cv_;
    };
 
    class OPENMS_DLLAPI ScopedSlot
    {
    public:
      explicit ScopedSlot(ConcurrencyLimiter* limiter);
      ~ScopedSlot();
 
      ScopedSlot(const ScopedSlot&) = delete;
      ScopedSlot& operator=(const ScopedSlot&) = delete;
 
    private:
      ConcurrencyLimiter* limiter_;
    };
  };
}