### Home Classes Methods

### In Files

- lib/taskjuggler/BatchProcessor.rb

### Parent

Object

### Methods

- ::new
- #queue
- #wait

### Files

- CHANGELOG
- COPYING
- README.rdoc

### Class/Module Index

Quicksearch

- TaskJuggler
- TaskJuggler::Account
- TaskJuggler::AccountAttribute
- TaskJuggler::AccountCredit
- TaskJuggler::AccountCreditListAttribute
- TaskJuggler::AccountListRE
- TaskJuggler::AccountScenario
- TaskJuggler::AlertLevelDefinition
- TaskJuggler::AlertLevelDefinitions
- TaskJuggler::Allocation
- TaskJuggler::AllocationAttribute
- TaskJuggler::AttributeBase
- TaskJuggler::AttributeDefinition
- TaskJuggler::AttributeOverwrite
- TaskJuggler::BatchProcessor
- TaskJuggler::Booking
- TaskJuggler::BookingListAttribute
- TaskJuggler::BooleanAttribute
- TaskJuggler::CSVFile
- TaskJuggler::CellSettingPattern
- TaskJuggler::CellSettingPatternList
- TaskJuggler::Charge
- TaskJuggler::ChargeListAttribute
- TaskJuggler::ChargeSet
- TaskJuggler::ChargeSetListAttribute
- TaskJuggler::ChartPlotter
- TaskJuggler::ChartPlotterError
- TaskJuggler::CollisionDetector
- TaskJuggler::ColumnListAttribute
- TaskJuggler::ColumnTable
- TaskJuggler::Daemon
- TaskJuggler::DaemonConnector
- TaskJuggler::DaemonConnectorMixin
- TaskJuggler::DataCache
- TaskJuggler::DataCacheEntry
- TaskJuggler::DateAttribute
- TaskJuggler::DefinitionListAttribute
- TaskJuggler::DependencyListAttribute
- TaskJuggler::DurationAttribute
- TaskJuggler::ExportRE
- TaskJuggler::FileList
- TaskJuggler::FileRecord
- TaskJuggler::FixnumAttribute
- TaskJuggler::FlagListAttribute
- TaskJuggler::FloatAttribute
- TaskJuggler::FormatListAttribute
- TaskJuggler::GanttChart
- TaskJuggler::GanttContainer
- TaskJuggler::GanttHeader
- TaskJuggler::GanttHeaderScaleItem
- TaskJuggler::GanttLine
- TaskJuggler::GanttLoadStack
- TaskJuggler::GanttMilestone
- TaskJuggler::GanttRouter
- TaskJuggler::GanttTaskBar
- TaskJuggler::HTMLDocument
- TaskJuggler::HTMLElements
- TaskJuggler::HTMLGraphics
- TaskJuggler::ICalReport
- TaskJuggler::ICalendar
- TaskJuggler::ICalendar::Component
- TaskJuggler::ICalendar::Event
- TaskJuggler::ICalendar::Journal
- TaskJuggler::ICalendar::Person
- TaskJuggler::ICalendar::Todo
- TaskJuggler::Interval
- TaskJuggler::IntervalList
- TaskJuggler::JobInfo
- TaskJuggler::Journal
- TaskJuggler::JournalEntry
- TaskJuggler::JournalEntryList
- TaskJuggler::JournalSortListAttribute
- TaskJuggler::KateSyntax
- TaskJuggler::KeywordArray
- TaskJuggler::KeywordDocumentation
- TaskJuggler::Leave
- TaskJuggler::LeaveAllowance
- TaskJuggler::LeaveAllowanceList
- TaskJuggler::LeaveAllowanceListAttribute
- TaskJuggler::LeaveList
- TaskJuggler::LeaveListAttribute
- TaskJuggler::Limits
- TaskJuggler::Limits::Limit
- TaskJuggler::LimitsAttribute
- TaskJuggler::ListAttributeBase
- TaskJuggler::Log
- TaskJuggler::LogicalAttribute
- TaskJuggler::LogicalExpression
- TaskJuggler::LogicalExpressionAttribute
- TaskJuggler::LogicalExpressionListAttribute
- TaskJuggler::LogicalFlag
- TaskJuggler::LogicalFunction
- TaskJuggler::LogicalOperation
- TaskJuggler::ManagerResponsibilities
- TaskJuggler::ManagerStatusRecord
- TaskJuggler::Message
- TaskJuggler::MessageHandler
- TaskJuggler::MessageHandlerInstance
- TaskJuggler::MspXmlRE
- TaskJuggler::Navigator
- TaskJuggler::NavigatorElement
- TaskJuggler::NikuProject
- TaskJuggler::NikuReport
- TaskJuggler::NikuResource
- TaskJuggler::NodeListAttribute
- TaskJuggler::PTNProxy
- TaskJuggler::Painter
- TaskJuggler::Painter::Circle
- TaskJuggler::Painter::Color
- TaskJuggler::Painter::Element
- TaskJuggler::Painter::Ellipse
- TaskJuggler::Painter::FontMetrics
- TaskJuggler::Painter::FontMetricsData
- TaskJuggler::Painter::Group
- TaskJuggler::Painter::Line
- TaskJuggler::Painter::Points
- TaskJuggler::Painter::PolyLine
- TaskJuggler::Painter::Primitives
- TaskJuggler::Painter::Rect
- TaskJuggler::Painter::SVGSupport
- TaskJuggler::Painter::Text
- TaskJuggler::PlaceHolderCell
- TaskJuggler::ProcessIntercom
- TaskJuggler::ProcessIntercomIface
- TaskJuggler::Project
- TaskJuggler::ProjectBroker
- TaskJuggler::ProjectBrokerIface
- TaskJuggler::ProjectFileParser
- TaskJuggler::ProjectFileScanner
- TaskJuggler::ProjectRecord
- TaskJuggler::ProjectServer
- TaskJuggler::ProjectServerIface
- TaskJuggler::PropertyAttribute
- TaskJuggler::PropertyList
- TaskJuggler::PropertySet
- TaskJuggler::PropertyTreeNode
- TaskJuggler::Query
- TaskJuggler::RTFHandlers
- TaskJuggler::RTFNavigator
- TaskJuggler::RTFQuery
- TaskJuggler::RTFReport
- TaskJuggler::RTFReportLink
- TaskJuggler::RTFWithQuerySupport
- TaskJuggler::RealFormat
- TaskJuggler::RealFormatAttribute
- TaskJuggler::ReferenceAttribute
- TaskJuggler::Report
- TaskJuggler::ReportBase
- TaskJuggler::ReportContext
- TaskJuggler::ReportScenario
- TaskJuggler::ReportServer
- TaskJuggler::ReportServerIface
- TaskJuggler::ReportServerRecord
- TaskJuggler::ReportServlet
- TaskJuggler::ReportTable
- TaskJuggler::ReportTableCell
- TaskJuggler::ReportTableColumn
- TaskJuggler::ReportTableLegend
- TaskJuggler::ReportTableLine
- TaskJuggler::Resource
- TaskJuggler::ResourceListAttribute
- TaskJuggler::ResourceListRE
- TaskJuggler::ResourceScenario
- TaskJuggler::RichText
- TaskJuggler::RichTextAttribute
- TaskJuggler::RichTextDocument
- TaskJuggler::RichTextElement
- TaskJuggler::RichTextFunctionExample
- TaskJuggler::RichTextFunctionHandler
- TaskJuggler::RichTextImage
- TaskJuggler::RichTextIntermediate
- TaskJuggler::RichTextParser
- TaskJuggler::RichTextScanner
- TaskJuggler::RichTextSnip
- TaskJuggler::RichTextSyntaxRules
- TaskJuggler::Scenario
- TaskJuggler::ScenarioData
- TaskJuggler::ScenarioListAttribute
- TaskJuggler::Scoreboard
- TaskJuggler::ScoreboardInterval
- TaskJuggler::SheetHandlerBase
- TaskJuggler::SheetReceiver
- TaskJuggler::SheetSender
- TaskJuggler::Shift
- TaskJuggler::ShiftAssignment
- TaskJuggler::ShiftAssignments
- TaskJuggler::ShiftAssignmentsAttribute
- TaskJuggler::ShiftScenario
- TaskJuggler::SimpleQueryExpander
- TaskJuggler::SortListAttribute
- TaskJuggler::StatusSheetReceiver
- TaskJuggler::StatusSheetReport
- TaskJuggler::StatusSheetSender
- TaskJuggler::StdIoWrapper
- TaskJuggler::StringAttribute
- TaskJuggler::SymbolAttribute
- TaskJuggler::SymbolListAttribute
- TaskJuggler::SyntaxReference
- TaskJuggler::TOCEntry
- TaskJuggler::TSResourceRecord
- TaskJuggler::TSTaskRecord
- TaskJuggler::TableColumnDefinition
- TaskJuggler::TableColumnSorter
- TaskJuggler::TableOfContents
- TaskJuggler::TableReport
- TaskJuggler::TableReportColumn
- TaskJuggler::TagFile
- TaskJuggler::TagFile::TagFileEntry
- TaskJuggler::Task
- TaskJuggler::TaskDepListAttribute
- TaskJuggler::TaskDependency
- TaskJuggler::TaskListAttribute
- TaskJuggler::TaskListRE
- TaskJuggler::TaskScenario
- TaskJuggler::TernarySearchTree
- TaskJuggler::TextFormatter
- TaskJuggler::TextParser
- TaskJuggler::TextParser::Macro
- TaskJuggler::TextParser::MacroTable
- TaskJuggler::TextParser::Pattern
- TaskJuggler::TextParser::Rule
- TaskJuggler::TextParser::Scanner
- TaskJuggler::TextParser::Scanner::BufferStreamHandle
- TaskJuggler::TextParser::Scanner::FileStreamHandle
- TaskJuggler::TextParser::Scanner::MacroStackEntry
- TaskJuggler::TextParser::Scanner::StreamHandle
- TaskJuggler::TextParser::SourceFileInfo
- TaskJuggler::TextParser::StackElement
- TaskJuggler::TextParser::State
- TaskJuggler::TextParser::StateTransition
- TaskJuggler::TextParser::TextParserResultArray
- TaskJuggler::TextParser::TokenDoc
- TaskJuggler::TextReport
- TaskJuggler::TimeInterval
- TaskJuggler::TimeIntervalListAttribute
- TaskJuggler::TimeSheet
- TaskJuggler::TimeSheetReceiver
- TaskJuggler::TimeSheetRecord
- TaskJuggler::TimeSheetReport
- TaskJuggler::TimeSheetSender
- TaskJuggler::TimeSheetSummary
- TaskJuggler::TimeSheets
- TaskJuggler::Tj3
- TaskJuggler::Tj3AppBase
- TaskJuggler::Tj3Client
- TaskJuggler::Tj3Daemon
- TaskJuggler::Tj3Man
- TaskJuggler::Tj3SheetAppBase
- TaskJuggler::Tj3SsReceiver
- TaskJuggler::Tj3SsSender
- TaskJuggler::Tj3TsReceiver
- TaskJuggler::Tj3TsSender
- TaskJuggler::Tj3TsSummary
- TaskJuggler::Tj3WebD
- TaskJuggler::TjException
- TaskJuggler::TjRuntimeError
- TaskJuggler::TjTime
- TaskJuggler::TjpExample
- TaskJuggler::TjpExportRE
- TaskJuggler::TjpSyntaxRules
- TaskJuggler::TraceReport
- TaskJuggler::URLParameter
- TaskJuggler::UserManual
- TaskJuggler::VimSyntax
- TaskJuggler::WebServer
- TaskJuggler::WelcomePage
- TaskJuggler::WorkingHours
- TaskJuggler::WorkingHoursAttribute
- TaskJuggler::XMLBlob
- TaskJuggler::XMLComment
- TaskJuggler::XMLDocument
- TaskJuggler::XMLElement
- TaskJuggler::XMLNamedText
- TaskJuggler::XMLText
- Diff
- Diff::Hunk
- AppConfig
- Diffable
- DiffableString
- Object
- RuntimeConfig
- String

No matching classes.

# TaskJuggler::BatchProcessor

The BatchProcessor class can be used to run code blocks of the program as a separate process. Mulitple pieces of code can be submitted to be executed in parallel. The number of CPU cores to use is limited at object creation time. The submitted jobs will be queued and scheduled to the given number of CPUs. The usage model is simple. Create an BatchProcessor object. Use BatchProcessor#queue to submit all the jobs and then use BatchProcessor#wait to wait for completion and to process the results.

### Public Class Methods

new(maxCpuCores) click to toggle source

Create a BatchProcessor object. `maxCpuCores` limits the number of simultaneously spawned processes.

    # File lib/taskjuggler/BatchProcessor.rb, line 67
    def initialize(maxCpuCores)
      @maxCpuCores = maxCpuCores
      # Jobs submitted by calling queue() are put in the @toRunQueue. The
      # pusher Thread will pick them up and fork them off into another
      # process.
      @toRunQueue = Queue.new
      # A hash that maps the JobInfo objects of running jobs by their PID.
      @runningJobs = { }
      # A list of jobs that wait to complete their writing.
      @spoolingJobs = [ ]
      # The wait() method will then clean the @toDropQueue, executes the post
      # processing block and removes all JobInfo related objects.
      @toDropQueue = Queue.new

      # A semaphore to guard accesses to @runningJobs, @spoolingJobs and
      # following shared data structures.
      @lock = Monitor.new
      # We count the submitted and completed jobs. The @jobsIn counter also
      # doubles as a unique job ID.
      @jobsIn = @jobsOut = 0
      # An Array that holds all the IO objects to receive data from.
      @pipes = []
      # A hash that maps IO objects to JobInfo objects
      @pipeToJob = {}

      # This global flag is set to true to signal the threads to terminate.
      @terminate = false
      # Sleep time of the threads when no data is pending. This value must be
      # large enough to allow for a context switch between the sending
      # (forked-off) process and this process. If it's too large, throughput
      # will suffer.
      @timeout = 0.02

      Thread.abort_on_exception = true
    end

### Public Instance Methods

queue(tag = nil, &block) click to toggle source

Add a new job the job queue. `tag` is some data that the caller can use to identify the job upon completion. `block` is a Ruby code block to be executed in a separate process.

    # File lib/taskjuggler/BatchProcessor.rb, line 106
    def queue(tag = nil, &block)
      raise 'You cannot call queue() while wait() is running!' if @jobsOut > 0

      # If this is the first queued job for this run, we have to start the
      # helper threads.
      if @jobsIn == 0
        # The JobInfo objects in the @toRunQueue are processed by the pusher
        # thread.  It forkes off processes to execute the code block associated
        # with the JobInfo.
        @pusher = Thread.new { pusher }
        # The popper thread waits for terminated childs and picks up the
        # results.
        @popper = Thread.new { popper }
        # The grabber thread collects $stdout and $stderr data from each child
        # process and stores them in the corresponding JobInfo.
        @grabber = Thread.new { grabber }
      end

      # Create a new JobInfo object for the job and push it to the @toRunQueue.
      job = JobInfo.new(@jobsIn, block, tag)
      # Increase job counter
      @lock.synchronize { @jobsIn += 1 }
      @toRunQueue.push(job)
    end

wait() click to toggle source

Wait for all jobs to complete. The code block will get the JobInfo objects for each job to pick up the results.

    # File lib/taskjuggler/BatchProcessor.rb, line 133
    def wait
      # Don't wait if there are no jobs.
      return if @jobsIn == 0

      # When we have received as many jobs in the @toDropQueue than we have
      # started then we're done.
      while !@lock.synchronize { @jobsIn == @jobsOut }
        if @toDropQueue.empty?
          sleep(@timeout)
        else
          # We have completed jobs.
          while !@toDropQueue.empty?
            # Pop a job from the @toDropQueue and call the block with it.
            job = @toDropQueue.pop
            # Remove the job related entries from the housekeeping tables.
            @lock.synchronize { @jobsOut += 1 }

            # Call the post-processing block that was passed to wait() with
            # the JobInfo object as argument.
            yield(job)
          end
        end
      end

      # Signal threads to stop
      @terminate = true
      # Wait for treads to finish
      @pusher.join
      @popper.join
      @grabber.join

      # Reset some variables so we can reuse the object for further job runs.
      @jobsIn = @jobsOut = 0
      @terminate = false

      # Make sure all data structures are empty and clean.
      check
    end

[Validate]

Generated with the Darkfish Rdoc Generator 2.