<html><head><meta name="color-scheme" content="light dark"></head><body><pre style="word-wrap: break-word; white-space: pre-wrap;">package TaskManager::Task::Queue;

use strict;
use warnings;

use TaskManager::Constants;
use TaskManager::DB;
use TaskManager::System;
use TaskManager::Task::Generic;

our @ISA = qw(TaskManager::Task::Generic);


sub new {
	my $class = shift || die "Bad method call!";
	my $self = $class-&gt;SUPER::new(@_);
	$self-&gt;{'table'} = 'task_queue';
	return $self;
}


sub allocate_task {
	my $self = shift || die "Bad method call!";
	my $WORKER_PID = shift or die "!";

	my $__gm = '';
#	$__gm .= ($__gm ? ',' : '').'?'
#		for( 1..scalar( @{$CFG-&gt;{'__MUTEX_GROUPS'}} ) );
#
#	my $sth = $dbh-&gt;prepare(&lt;&lt;SQL) or make_die($dbh-&gt;errstr);

	my $SQL = &lt;&lt;SQL;
		SELECT
			id, exec_str, UNIX_TIMESTAMP(dt_start) AS dt_start_uts, priority
		FROM
			task_queue
		WHERE
			(
				worker_status=? OR
				(
					worker_status=? AND
					repeat_on_error!=0 AND
					attempt_no*$CFG-&gt;{NEXT_ATTEMPT_TIMEOUT} + UNIX_TIMESTAMP(dt_start) &lt; UNIX_TIMESTAMP(NOW())
				)
			)
--			AND
--			gmutex IN ($__gm)
		ORDER BY
			priority ASC,
			attempt_no ASC,
			dt_start ASC
		LIMIT
			0,1
SQL
#
#	$sth-&gt;execute(TM_STATUS_NEW, TM_STATUS_FAIL, @{$CFG-&gt;{'__MUTEX_GROUPS'}}) or make_die($sth-&gt;errstr);
#	return (undef,undef)
#		unless $sth-&gt;rows();

	my $REZ = TaskManager::DB::select_query(
		$SQL,
		TM_STATUS_NEW, TM_STATUS_FAIL
	);


	return undef
		unless scalar @$REZ;

	my $task_data = pop @$REZ;

	unless( $task_data-&gt;{'exec_str'} ) {
		$self-&gt;set_task_error($task_data-&gt;{'id'}, 0xFF, 'illegal params');
		return undef;
	}


	## looking for free slots (if reserve defined)
	if( $task_data-&gt;{'priority'}&gt;1 ) {

		my $total_low_pr = scalar grep { $WORKER_PID-&gt;{$_}-&gt;[2] &gt; 1 } (keys %$WORKER_PID);

		return (undef,undef,undef)
			if( $CFG-&gt;{'RESERVED_HIGH_PRIORITY_PROCESS'}
				&amp;&amp; $CFG-&gt;{'MAX_CHILD_PROCESS'} - $total_low_pr &lt;= $CFG-&gt;{'RESERVED_HIGH_PRIORITY_PROCESS'}
			);
	}


	my $task_exec_str = $self-&gt;parse_exec_str($task_data-&gt;{'exec_str'});

	unless( $self-&gt;validate_exec_str($task_exec_str) ) {
		$self-&gt;set_task_error($task_data-&gt;{'id'}, 0xFF, 'illegal params');
		return undef;
	}

	TaskManager::DB::do_query(
		"UPDATE task_queue SET dt_start=NOW(), attempt_no=attempt_no+1 WHERE id=?",
		$task_data-&gt;{'id'}
	)
		or make_die($TaskManager::DB::error);

	return ( $task_data-&gt;{'id'}, [ split(/ +/, $task_exec_str) ], $task_data-&gt;{'priority'} );
}


sub check_server {

	my $REZ = TaskManager::DB::select_query(
		"SELECT * FROM server_list WHERE code=?",
		$CFG-&gt;{'SERVER_CODE'}
	);

	die "No server found by code '$CFG-&gt;{SERVER_CODE}'!"
		unless scalar @$REZ;

	die "Collision detected! More that one server found!"
		if ( scalar(@$REZ) &gt; 1 );

	my $row = pop @$REZ;

	die "Server isn't active!"
		if $row-&gt;{'status'} ne TM_SERVER_STATUS_ACTIVE;

	die "Server has illegal limit type!"
		if( $row-&gt;{'limit_type'} ne TM_SERVER_LIMIT_TYPE_LOADAVG
			&amp;&amp; $row-&gt;{'limit_type'} ne TM_SERVER_LIMIT_TYPE_TASKNUM
		);

	die "No task max defined for queue server!"
		if( $row-&gt;{'limit_type'} eq TM_SERVER_LIMIT_TYPE_TASKNUM &amp;&amp; !$row-&gt;{'max_tasknum'} );

	if( !$row-&gt;{'max_tasknum'} || $row-&gt;{'max_tasknum'}&gt;TM_MAX_TOTAL_CHILD ) {
		drop_log("Max task num. set to ".TM_MAX_TOTAL_CHILD." from ".$row-&gt;{'max_tasknum'});
		$row-&gt;{'max_tasknum'} = TM_MAX_TOTAL_CHILD;
	}

	$CFG-&gt;{'SERVER_TYPE'} = $row-&gt;{'limit_type'};
	$CFG-&gt;{'SERVER_ID'} = $row-&gt;{'id'};
	$CFG-&gt;{'MAX_CHILD_PROCESS'} = $row-&gt;{'max_tasknum'};

	return undef;
}


## now it checks the task for current server (not for the whole project)
sub check_uncompleted_tasks {

#	my $__gm = '';
#	$__gm .= ($__gm ? ',' : '').'?'
#		for( 1..scalar( @{$CFG-&gt;{'__MUTEX_GROUPS'}} ) );

#	my $sth = $dbh-&gt;prepare() or make_die($dbh-&gt;errstr);
#	$sth-&gt;execute(TM_STATUS_WORK, @{$CFG-&gt;{'__MUTEX_GROUPS'}}) or make_die($sth-&gt;errstr);

	my $uncomp_tasks = TaskManager::DB::select_query(
		"SELECT COUNT(*) AS total_uncomp FROM task_queue WHERE worker_status=? AND id_worker_server=?",
		TM_STATUS_WORK, $CFG-&gt;{'SERVER_ID'}
	);

	drop_log("WARNING! ".$uncomp_tasks-&gt;[0]-&gt;{'total_uncomp'}." uncompleted tasks found for server $CFG-&gt;{SERVER_CODE}!")
		if ( scalar @$uncomp_tasks &amp;&amp; $uncomp_tasks-&gt;[0]-&gt;{'total_uncomp'}&gt;0 );

	return undef;
}


1;
</pre></body></html>