### Create and Manage MCE Queues

Source: https://context7.com/marioroy/mce-perl/llms.txt

Demonstrates how to create priority queues with custom settings (order and type), enqueue and dequeue items with and without priorities, and manage queue status. It also includes a basic producer-consumer example using MCE.

```perl
use MCE;

# Create priority queue with custom settings
my $pqueue = MCE::Queue->new(
    porder => $MCE::Queue::HIGHEST,  # or $MCE::Queue::LOWEST
    type => $MCE::Queue::FIFO        # or $MCE::Queue::LIFO
);

# Enqueue items
$queue->enqueue('item1', 'item2', 'item3');
$pqueue->enqueuep(5, 'high priority item');  # with priority

# Dequeue items
my $item = $queue->dequeue();              # blocking
my $item_nb = $queue->dequeue_nb();        # non-blocking
my $item_timed = $queue->dequeue_timed(5); # 5 second timeout

# Check queue status
my $count = $queue->pending();
$queue->clear();
$queue->end();  # Signal end of queue

# Producer-consumer example
my $mce = MCE->new(
    max_workers => 4,
    user_func => sub {
        while (defined(my $item = $queue->dequeue())) {
            # Process item
            print "Worker " . MCE->wid . " processing: $item\n";
        }
    }
);

# Producer
for (1..100) { $queue->enqueue($_); }
$queue->end();

$mce->run();
```

--------------------------------

### MCE Core API and Flow Model Initialization (Perl)

Source: https://github.com/marioroy/mce-perl/blob/master/README.md

Demonstrates two ways to initialize MCE: using the Core API with a direct 'new' call and using the MCE::Flow model for a more concise syntax. Both examples set the maximum number of workers and define a user function to be executed.

```perl
use MCE;

my $mce = MCE->new(
   max_workers => 5,
   user_func => sub {
      my ($mce) = @_;
      $mce->say("Hello from " . $mce->wid);
   }
);

$mce->run;

# Construction using a MCE model

use MCE::Flow max_workers => 5;

mce_flow sub {
   my ($mce) = @_;
   MCE->say("Hello from " . MCE->wid);
 };
```

--------------------------------

### MCE Channel - Bidirectional Communication

Source: https://context7.com/marioroy/mce-perl/llms.txt

Illustrates using MCE::Channel for efficient bidirectional communication between workers. Covers creating channels with different implementations (Mutex, SimpleFast), sending and receiving data (blocking and non-blocking), and a multi-worker communication example.

```perl
use MCE::Channel;

# Create channel with default implementation (Mutex or Threads)
my $channel = MCE::Channel->new();

# Create fast channel (lock-free for single producer/consumer)
my $fast_channel = MCE::Channel->new(impl => 'SimpleFast');

# Create channel with specific implementation
my $chan = MCE::Channel->new(impl => 'Mutex');
# Options: Mutex, MutexFast, Simple, SimpleFast, Threads, ThreadsFast

# Send data through channel
$channel->send('message1');
$channel->send2('message2');  # Alternative send method

# Receive data from channel
my $msg = $channel->recv();       # blocking
my $msg_nb = $channel->recv_nb(); # non-blocking

# Send and receive multiple items
$channel->send('item1', 'item2', 'item3');
my @items = $channel->recv();

# Check implementation
print "Channel impl: " . $channel->impl() . "\n";

# Multi-worker communication example
use MCE;

my $input_chan = MCE::Channel->new();
my $output_chan = MCE::Channel->new();

my $mce = MCE->new(
    max_workers => 3,
    user_func => sub {
        while (defined(my $data = $input_chan->recv())) {
            my $result = process_data($data);
            $output_chan->send($result);
        }
    }
);

$mce->spawn();

# Producer
for (1..100) { $input_chan->send($_); }
$input_chan->end();

# Consumer
while (defined(my $result = $output_chan->recv())) {
    print "Result: $result\n";
}

$mce->shutdown();
```

--------------------------------

### MCE Parallel Flow: General Purpose Parallelization (Perl)

Source: https://context7.com/marioroy/mce-perl/llms.txt

Introduces MCE::Flow for executing general-purpose parallel workflows. It covers initializing workers and setting options like `bounds_only`, and provides an example of computing Pi using parallel flow with sequence processing. Requires MCE::Flow and cleanup.

```perl
use MCE::Flow;

MCE::Flow->init(
    max_workers => 8,
    chunk_size => 200_000,
    bounds_only => 1  # Workers receive [begin, end] bounds
);

# Compute pi using parallel flow
my $N = 4_000_000;

sub compute_pi {
    my ($beg_seq, $end_seq) = @_;
    my $pi = 0.0;

    foreach my $i ($beg_seq .. $end_seq) {
        my $t = ($i + 0.5) / $N;
        $pi += 4.0 / (1.0 + $t * $t);
    }

    MCE->gather($pi);
}

my @results = mce_flow_s sub {
    compute_pi($_->[0], $_->[1]);
}, 0, $N - 1;

my $pi = 0.0;
$pi += $_ for @results;
printf "pi = %0.13f\n", $pi / $N;  # 3.1415926535898

MCE::Flow->finish;
```

--------------------------------

### MCE Mutex - Cross-Platform Locking

Source: https://context7.com/marioroy/mce-perl/llms.txt

Shows how to use MCE::Mutex for cross-platform synchronization primitives to protect shared resources across workers. Examples include creating in-memory and file-based mutexes, locking/unlocking, and using synchronize for critical sections.

```perl
use MCE::Flow max_workers => 4;
use MCE::Mutex;

# Create mutex for synchronization
my $mutex = MCE::Mutex->new();

# Create file-based mutex
my $file_mutex = MCE::Mutex->new(
    impl => 'Flock',
    path => '/tmp/my.lock'
);

# Protect shared resource
my $counter = 0;

mce_flow sub {
    for (1..1000) {
        $mutex->lock;
        $counter++;
        $mutex->unlock;
    }
};

print "Counter: $counter\n";  # 4000

# Synchronize code block
$mutex->synchronize(sub {
    # Critical section
    $counter += 10;
});

# Try to lock (non-blocking)
if ($mutex->lock_exclusive()) {
    # Got lock
    $mutex->unlock();
}

# Lock with shared/exclusive modes
$mutex->lock_shared();    # Multiple readers
$mutex->unlock_shared();

MCE::Flow->finish;
```

--------------------------------

### MCE Core API: Construct and Run Parallel Workers in Perl

Source: https://context7.com/marioroy/mce-perl/llms.txt

Demonstrates creating and managing parallel workers using MCE's core API. It covers basic worker pool setup, defining user functions for processing, and running workers with or without input data. This is foundational for understanding MCE's parallel execution model.

```perl
use MCE;

# Basic worker pool with 5 workers
my $mce = MCE->new(
    max_workers => 5,
    user_func => sub {
        my ($mce) = @_;
        my $wid = $mce->wid;  # worker ID
        $mce->say("Worker $wid processing");

        # Process data chunks
        my $chunk = $mce->chunk;
        foreach my $item (@$chunk) {
            # Process each item
        }
    }
);

# Run the workers
$mce->run;

# Run with input data
my @data = 1..1000;
$mce->process({ input_data => \@data });
```

--------------------------------

### MCE Parallel Grep: Filter Data in Parallel (Perl)

Source: https://context7.com/marioroy/mce-perl/llms.txt

Details the usage of MCE::Grep for parallel data filtering. Examples include filtering arrays, files with pattern matching, and sequences, all while preserving order. Requires MCE::Grep and a cleanup step with MCE::Grep->finish.

```perl
use MCE::Grep;

MCE::Grep->init(
    max_workers => 8,
    chunk_size => 500
);

# Parallel grep over array
my @data = 1..10000;
my @even = mce_grep { $_ % 2 == 0 } @data;

# Parallel grep over file with pattern matching
my @matches = mce_grep_f {
    chomp;
    /ERROR/ || /FATAL/
} '/var/log/application.log';

# Parallel grep over sequence
my @primes = mce_grep_s {
    my ($mce, $n, $chunk_id) = @_;
    return 0 if $n < 2;
    for (2..sqrt($n)) { return 0 if $n % $_ == 0; }
    return 1;
} 1, 1000;

MCE::Grep->finish;
```

--------------------------------

### MCE Worker Control Methods in Perl

Source: https://context7.com/marioroy/mce-perl/llms.txt

Control worker behavior and retrieve worker-specific information during parallel execution using MCE methods. Includes methods for getting worker IDs, outputting messages, gathering results, sending data, and managing execution flow.

```perl
use MCE;

my $mce = MCE->new(
    max_workers => 4,
    chunk_size => 100,
    user_func => sub {
        my ($mce) = @__;

        # Get worker information
        my $wid = $mce->wid;           # Worker ID (1..N)
        my $task_id = $mce->task_id;   # Task ID
        my $task_wid = $mce->task_wid; # Worker ID within task
        my $chunk_id = $mce->chunk_id; # Current chunk ID

        # Output methods
        $mce->say("Message with newline");
        $mce->print("Message without newline");
        $mce->printf("Formatted: %d\n", $wid);

        # Gather results
        $mce->gather($result);
        $mce->gather($chunk_id, \@results);

        # Send data to specific worker or task
        $mce->sendto('worker:3', $data);
        $mce->sendto('task:2', $data);

        # Abort processing
        $mce->abort() if $error_condition;

        # Exit worker
        $mce->exit(0, 'optional message');

        # Relay data between workers
        $mce->relay_recv();  # Receive from previous worker
        $mce->relay($data);  # Send to next worker

        # Process next chunk
        $mce->do_next();

        # Last iteration?
        if ($mce->last_iteration()) {
            # Cleanup
        }
    }
);

# Configuration accessors
my $max = $mce->max_workers();
my $chunk = $mce->chunk_size();
my $args = $mce->user_args();
my $task_name = $mce->task_name();

# Freeze/thaw (serialization)
my $frozen = $mce->freeze($data);
my $data = $mce->thaw($frozen);

$mce->run();
```

--------------------------------

### Parallel Log File Parsing with MCE::Loop (Perl)

Source: https://github.com/marioroy/mce-perl/blob/master/README.md

Shows how to use MCE::Loop to parse a large file in parallel. It initializes MCE with slurpio enabled for efficient file reading and demonstrates how to search for a pattern within each chunk. Includes platform-specific optimizations for Windows.

```perl
use MCE::Loop;

MCE::Loop->init( max_workers => 8, use_slurpio => 1 );

my $pattern  = 'something';
my $hugefile = 'very_huge.file';

my @result = mce_loop_f {
   my ($mce, $slurp_ref, $chunk_id) = @_;

   # Quickly determine if a match is found.
   # Process the slurped chunk only if true.

   if ($$slurp_ref =~ /$pattern/m) {
      my @matches;

      # The following is fast on Unix, but performance degrades
      # drastically on Windows beyond 4 workers.

      open my $MEM_FH, '<', $slurp_ref;
      binmode $MEM_FH, ':raw';
      while (<$MEM_FH>) { push @matches, $_ if (/$pattern/); }
      close   $MEM_FH;

      # Therefore, use the following construction on Windows.

      while ( $$slurp_ref =~ /([^
]+\n)/mg ) {
         my $line = $1; # save $1 to not lose the value
         push @matches, $line if ($line =~ /$pattern/);
      }

      # Gather matched lines.

      MCE->gather(@matches);
   }

 } $hugefile;

 print join('', @result);
```

--------------------------------

### MCE Queue: Thread-Safe Data Queue (Perl)

Source: https://context7.com/marioroy/mce-perl/llms.txt

Demonstrates the creation and basic usage of MCE::Queue for thread-safe data queuing, suitable for producer-consumer patterns. It shows how to instantiate a queue with default settings using the MCE module. Further operations like enqueue/dequeue are implied.

```perl
use MCE;
use MCE::Queue;

# Create queue with default settings
my $queue = MCE::Queue->new();

```

--------------------------------

### Configure MCE with Advanced Perl Options

Source: https://context7.com/marioroy/mce-perl/llms.txt

This snippet demonstrates how to initialize and configure an MCE object in Perl with a wide array of advanced options. It covers worker configuration, input data sources, gather/reduce functions, various callbacks for lifecycle management, output redirection, and serialization.

```perl
use MCE;

my $mce = MCE->new(
    # Worker configuration
    max_workers => 'auto',        # or integer, '50%', '1.5'
    chunk_size => 'auto',         # or integer
    use_threads => 0,             # Force process mode
    spawn_delay => 0.1,           # Delay between worker spawns
    submit_delay => 0.01,         # Delay between chunk submissions
    job_delay => 0.001,           # Delay between jobs

    # Input data sources
    input_data => \@array,        # Array reference
    # OR
    input_data => $file_handle,   # File handle
    # OR
    input_data => \&iterator,     # Iterator function
    # OR
    sequence => { begin => 1, end => 1000, step => 1 },

    # Input options
    RS => "\n",                   # Record separator
    bounds_only => 0,             # Send bounds instead of data
    use_slurpio => 0,             # Slurp file chunks

    # Gather/reduce results
    gather => \&gather_function,

    # Callbacks
    user_begin => sub {
        my ($mce, $task_id, $task_name) = @_;
        # Called once per worker at start
    },
    user_end => sub {
        my ($mce, $task_id, $task_name) = @_;
        # Called once per worker at end
    },
    user_func => sub {
        my ($mce, $chunk_ref, $chunk_id) = @_;
        # Main processing function
    },
    user_error => sub {
        my ($mce, $msg, $wid) = @_;
        # Error handler
    },
    user_output => sub {
        my ($mce, @output) = @_;
        # Custom output handler
    },

    # Lifecycle hooks
    on_post_exit => sub {
        my ($mce, $e) = @_;
        # After worker exits
    },
    on_post_run => sub {
        my ($mce, $task_id, $task_name) = @_;
        # After run completes
    },

    # Output redirection
    stdout_file => '/path/to/stdout.log',
    stderr_file => '/path/to/stderr.log',
    flush_stdout => 1,
    flush_stderr => 1,
    flush_file => 1,

    # Advanced options
    parallel_io => 1,             # Enable parallel I/O
    max_retries => 0,             # Retry failed chunks
    loop_timeout => 0,            # Timeout for loops
    interval => 0,                # Callback interval
    progress => \&progress_func,  # Progress callback
    posix_exit => 0,              # Use POSIX::_exit

    # Serialization
    freeze => \&custom_freeze,
    thaw => \&custom_thaw,

    # Relay communication
    init_relay => 0,              # or initial value

    # Multi-task configuration
    user_tasks => [{
        max_workers => 2,
        task_name => 'task1',
        user_func => \&task1_func,
        sequence => { begin => 1, end => 100 }
    }, {
        max_workers => 3,
        task_name => 'task2',
        user_func => \&task2_func,
        input_data => \@data2
    }],

    # Temporary directory
    tmp_dir => '/tmp/mce',

    # User arguments (accessible via $mce->user_args)
    user_args => { key => 'value' }
);

# Run with override options
$mce->run({ max_workers => 8 }, sub {
    # Process function
});

# Process with different input
$mce->process({ input_data => \@new_data });

# Spawn workers without running
$mce->spawn();
# ... do other work ...
$mce->shutdown();

```

--------------------------------

### MCE Parallel Map: Transform Data in Parallel (Perl)

Source: https://context7.com/marioroy/mce-perl/llms.txt

Illustrates using MCE::Map for parallel data transformations. It shows initialization, applying map operations to arrays, files, and sequences, and the automatic ordering of results. Requires MCE::Map module and includes cleanup with MCE::Map->finish.

```perl
use MCE::Map;

# Initialize with 4 workers
MCE::Map->init(
    max_workers => 4,
    chunk_size => 100
);

# Parallel map over array
my @input = 1..1000;
my @squared = mce_map { $_ * $_ } @input;

# Parallel map over file
my @results = mce_map_f {
    chomp;
    uc($_)  # uppercase each line
} '/path/to/file.txt';

# Parallel map over sequence
my @computed = mce_map_s {
    my ($mce, $num, $chunk_id) = @_;
    sqrt($num)
} 1, 10000;

# Cleanup
MCE::Map->finish;
```

--------------------------------

### MCE Child - Process Management API

Source: https://context7.com/marioroy/mce-perl/llms.txt

Details the MCE::Child module for spawning and managing child processes with a threads-like interface, compatible with Perl 5.8+. Includes initialization, creating children, waiting for them (blocking/non-blocking), and process control (kill, is_running).

```perl
use MCE::Child;

# Initialize with configuration
MCE::Child->init(
    max_workers => 4,
    child_timeout => 30,
    posix_exit => 1,
    on_start => sub {
        my ($pid, $ident) = @_;
        print "Child $ident started\n";
    },
    on_finish => sub {
        my ($pid, $exit, $ident, $signal, $error) = @_;
        print "Child $ident finished with exit code $exit\n";
    }
);

# Create child processes
my $child1 = MCE::Child->create(sub {
    my (@args) = @_;
    # Child process code
    sleep 2;
    return "result from child";
}, $arg1, $arg2);

# Create multiple children
my @children;
for (1..4) {
    push @children, MCE::Child->create('work_function', $_);
}

sub work_function {
    my ($id) = @_;;
    print "Working on $id\n";
    return $id * 2;
}

# Wait for child and get result
my $result = $child1->join();
print "Child returned: $result\n";

# Wait for all children
MCE::Child->wait_all();

# Non-blocking wait
my @finished = MCE::Child->wait_one();

# Kill child process
$child1->kill('TERM');

# Check if child is running
if ($child1->is_running()) {
    print "Child still running\n";
}

# Get child PID
my $pid = $child1->pid();

# Detach child
$child1->detach();

# Self reference in child
my $self = MCE::Child->self();

# List all children
my @list = MCE::Child->list();

MCE::Child->finish();
```

--------------------------------

### Parallel Computation of Pi using MCE::Flow (Perl)

Source: https://github.com/marioroy/mce-perl/blob/master/README.md

Illustrates using MCE::Flow to parallelize the computation of Pi. It defines a 'compute_pi' subroutine and then uses MCE::Flow with 'bounds_only' enabled to distribute the computation across workers. The results are then aggregated.

```perl
use MCE::Flow;

my $N = shift || 4_000_000;

sub compute_pi {
   my ( $beg_seq, $end_seq ) = @_;
   my ( $pi, $t ) = ( 0.0 );

   foreach my $i ( $beg_seq .. $end_seq ) {
      $t = ( $i + 0.5 ) / $N;
      $pi += 4.0 / ( 1.0 + $t * $t );
   }

   MCE->gather( $pi );
}

# Compute bounds only, workers receive [ begin, end ] values

MCE::Flow->init(
   chunk_size  => 200_000,
   max_workers => 8,
   bounds_only => 1
);

my @ret = mce_flow_s sub {
   compute_pi( $_->[0], $_->[1] );
 }, 0, $N - 1;

my $pi = 0.0;  $pi += $_ for @ret;

printf "pi = %0.13f\n", $pi / $N;  # 3.1415926535898
```

--------------------------------

### MCE Parallel Loop: Custom Logic Parallel Processing (Perl)

Source: https://context7.com/marioroy/mce-perl/llms.txt

Explains MCE::Loop for executing parallel loops with custom processing logic and data gathering. It demonstrates file parsing with slurping and processing arrays, including how to use MCE->gather for collecting results. Requires MCE::Loop and cleanup.

```perl
use MCE::Loop;

MCE::Loop->init(
    max_workers => 8,
    chunk_size => 1000,
    use_slurpio => 1  # Enable file slurping
);

# Parse huge log file in parallel
my $pattern = 'ERROR';
my @errors = mce_loop_f {
    my ($mce, $slurp_ref, $chunk_id) = @_;

    # Process only if pattern exists in chunk
    if ($$slurp_ref =~ /$pattern/m) {
        my @matches;
        while ($$slurp_ref =~ /([^\n]+\n)/mg) {
            my $line = $1;
            push @matches, $line if $line =~ /$pattern/;
        }
        MCE->gather(@matches);
    }
} '/var/log/huge_file.log';

# Loop over array with gathering
my $total = 0;
mce_loop {
    my ($mce, $chunk_ref, $chunk_id) = @_;
    my $sum = 0;
    $sum += $_ for @$chunk_ref;
    MCE->gather($sum);
} \@data;

MCE::Loop->finish;
```

--------------------------------

### MCE::Step Multi-Stage Processing in Perl

Source: https://context7.com/marioroy/mce-perl/llms.txt

Execute multi-stage parallel workflows where output from one stage feeds into the next using MCE::Step. Allows defining pipelines with varying worker counts per stage and sequential processing steps.

```perl
use MCE::Step;

# Define multi-step pipeline
MCE::Step->init(
    max_workers => [2, 4, 3],  # Workers per step
    bounds_only => 1,
    user_func => [&step1, &step2, &step3]
);

sub step1 {
    my ($mce, $chunk_ref, $chunk_id) = @__;
    my @results;
    for my $item (@$chunk_ref) {
        push @results, process_step1($item);
    }
    MCE->gather(MCE->task_wid, @results);
}

sub step2 {
    my ($mce, $chunk_ref, $chunk_id) = @__;
    my @results;
    for my $item (@$chunk_ref) {
        push @results, process_step2($item);
    }
    MCE->gather(MCE->task_wid, @results);
}

sub step3 {
    my ($mce, $chunk_ref, $chunk_id) = @__;
    for my $item (@$chunk_ref) {
        print "Final: $item\n";
    }
}

# Run the pipeline
my @input = 1..1000;\nmce_step { $_ } @input;

# Step with sequence processing
mce_step_s sub {
    my ($mce, $n, $chunk_id) = @__;
    my $result = expensive_computation($n);
    MCE->sendto('next_task', $result);
}, 0, 999;

MCE::Step->finish;
```

--------------------------------

### MCE::Stream Pipeline Processing in Perl

Source: https://context7.com/marioroy/mce-perl/llms.txt

Chain multiple map and grep operations into efficient parallel processing pipelines using MCE::Stream. Supports processing data in memory or from files, with support for multi-stage pipelines.

```perl
use MCE::Stream;

MCE::Stream->init(
    max_workers => 4,
    chunk_size => 1000,
    default_mode => 'map'  # or 'grep'
);

# Stream with multiple operations
my @results = mce_stream {
    # First stage: transform
    my ($mce, $chunk_ref, $chunk_id) = @__;
    my @transformed = map { $_ * 2 } @$chunk_ref;
    MCE->say("Stage 1: Worker " . MCE->wid);
    \@transformed;
} 1..1000;

# Stream over file
my @processed = mce_stream_f {
    chomp;
    my $line = $_;
    # Process line
    return $line if $line =~ /pattern/;
    return ();
} '/path/to/input.txt';

# Multi-stage pipeline
MCE::Stream->init(
    user_tasks => [{
        max_workers => 2,
        task_name => 'filter',
        user_func => sub {
            my ($mce, $chunk_ref, $chunk_id) = @__;
            my @filtered = grep { $_ > 100 } @$chunk_ref;
            MCE->gather($chunk_id, \@filtered);
        }
    }, {
        max_workers => 2,
        task_name => 'transform',
        user_func => sub {
            # Second stage processing
        }
    }]
);

MCE::Stream->finish;
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.