taskman_atf/atf.inc.php

<?php
namespace taskman;
use Amp;

interface IATFDevicePool
{
  function get();
}

class ATFApkInstaller
{
  const MAX_INSTALLS_IN_PROGRESS = 5;

  const ST_IN_PROGRESS = 1;
  const ST_INSTALLED   = 2;

  private $apk_name;
  private $installs = array();
  private $override_conf;

  function __construct($apk_name, $override_conf)
  {
    $this->apk_name = $apk_name;
    $this->override_conf = $override_conf;
  }

  function countInstallsInProgress()
  {
    $c = 0;
    foreach($this->installs as $status)
      if($status == self::ST_IN_PROGRESS)
        ++$c;
    return $c;
  }

  function installAsync($device)
  {
    return Amp\call(function() use($device) {
    
    while($this->countInstallsInProgress() >= self::MAX_INSTALLS_IN_PROGRESS)
      yield Amp\delay(1000);

    if(isset($this->installs[$device]))
      return;
    $this->installs[$device] = self::ST_IN_PROGRESS;
    try
    {
      yield atf_host_exec_async("%{adb}% -s $device shell am force-stop %{package_id}%", DEPLOY_OPT_ERR_OK);
      if($this->apk_name !== null)
      {
        yield atf_host_exec_async("%{adb}% -s $device uninstall %{package_id}%", DEPLOY_OPT_ERR_OK, 30);
        yield atf_host_exec_async("%{adb}% -s $device install -r ./{$this->apk_name}", 0, 400);
      }

      $this->installs[$device] = self::ST_INSTALLED;
    }
    catch(Exception $e)
    {
      atf_log("Error during install *$device*: " . $e->getMessage());
      unset($this->installs[$device]);
      throw $e;
    }
    });
  }
}


class ATFAdbDevicePool implements IATFDevicePool
{
  function get()
  {
    return atf_get_devices();
  }
}


class ATFFixedDevicePool implements IATFDevicePool
{
  private $devices;

  function __construct(array $devices)
  {
    $this->devices = $devices;
  }

  function get()
  {
    return $this->devices;
  }
}


class ATFCachedDevices implements IATFDevicePool
{
  private $provider;
  private $cached = null;
  private $last_cache_time;
  private $keep_cache_time;

  function __construct(IATFDevicePool $provider, $keep_cache_time)
  {
    $this->provider = $provider;
    $this->keep_cache_time = $keep_cache_time;
  } 

  function get()
  {
    if($this->cached === null || (time() - $this->last_cache_time) > $this->keep_cache_time)
    {
      $this->cached = $this->provider->get();
      $this->last_cache_time = time();
    }
    return $this->cached;
  }
}



class ATFSession
{
  public $guid;
  public $version;
  public $branch;
  public $rev_hash;

  public $plans = array();

  public $ignored_devices = array();
  public $bogus_device_count = array();

  public $share_with_qa_chan = true;
  public $shared_qa_errors = array();

  function __construct()
  {
    $this->guid = atf_guid();
    $this->branch = taskman_prop('GAME_VERSION_BRANCH'); 
    $this->version = taskman_prop('GAME_VERSION');
    $this->rev_hash = taskman_prop('GAME_REVISION_HASH');
  }

  function incBogusDevice($device)
  {
    if(!isset($this->bogus_device_count[$device]))
      $this->bogus_device_count[$device] = 0;
    return ++$this->bogus_device_count[$device];
  }

  function resetBogusDeviceCount($device)
  {
    $this->bogus_device_count[$device] = 0;
  }

  function resetBogusDevices()
  {
    $this->bogus_device_count = array();
  }

  function addPlan(ATFPlan $plan)
  {
    $plan->session = $this;
    $this->plans[] = $plan;
  }

  function resetPlans()
  {
    $this->plans = array();
  }

  function run($apk_path = null, $apk_reuse = false, $override_conf = false, $adb_reboot = true)
  {
    $apk_name = $apk_path === null ? null : basename($apk_path);

    _atf_start_watchdog($this->guid);

    //1. deploy an apk file to the atf host
    if(!$apk_reuse && $apk_path !== null)
    {
      if(!is_file($apk_path))
        throw new Exception("No such file '$apk_path'");

      $local_crc = _atf_get_local_file_crc($apk_path);
      $remote_crc = _atf_get_remote_file_crc($apk_name);

      if($remote_crc === '' || $local_crc !== $remote_crc)
        atf_host_put_file($apk_path, $apk_name);
      else
        atf_log("Skipping same .apk upload");
    }

    //2. reboot just in case
    if($adb_reboot)
      atf_adb_reboot();


    $install = new ATFApkInstaller($apk_name, $override_conf);
    foreach($this->plans as $plan)
      $this->_runPlan($install, $plan);

    foreach($this->plans as $plan)
    {
      if($plan->hasFatalProblems() ||
        ($plan->isOver() && !$plan->getDevices()))
        return false;
    }

    return true;
  }


  function _runPlan(ATFApkInstaller $install, ATFPlan $plan)
  {
    atf_log("Starting " . $plan->getTitle() . "...");

    //NOTE: hung_threshold must be larger than gone_threshold
    Amp\Promise\wait($plan->runAsync($install, $sleep_time = 4, $hung_threshold = 80, $gone_threshold = 30, $stuck_threshold = 200));
  }

  // function trySendStatsFromJzonAsync(ATFTask $task, $jzon)
  // {
  //   return Amp\call(function() use($task, $jzon) {

  //   try
  //   {
  //     $data = jzon_parse(trim(str_replace('\"', '"', $jzon)));
  //     $table = $data['table'];
  //     unset($data['table']);

  //     if(isset($data['deviceMemoryUsage']) && $data['deviceMemoryUsage'] === '')
  //     {
  //       $mem = yield atf_device_mem_async($task->device);
  //       $data['deviceMemoryUsage'] = $mem['total'];
  //       if($table === 'device_memory')
  //       {
  //         $data['deviceMemoryUsageNative'] = $mem['native'];
  //         $data['deviceMemoryUsageSystem'] = $mem['system'];
  //         $data['deviceMemoryUsageJava'] = $mem['java'];
  //         $data['deviceMemoryUsageGraphics'] = $mem['graphics'];
  //       }
  //     }
  //     $this->trySendStats($task, $table, $data);
  //   }
  //   catch(Exception $e)
  //   {
  //     echo $e->getMessage() . "\n";
  //   }

  //   });
  // }

// function trySendApkStatsEvent()
// {
//   try
//   {
//     $apk_sizeB = get_apk_size();
//     $apk_sizeMb = round($apk_sizeB / 1024 / 1024,4);
//     $version = taskman_prop('GAME_VERSION');
//     $rev_hash = taskman_prop('GAME_REVISION_HASH');
//     $msg = "version $version($rev_hash) size: $apk_sizeMb Mb ($apk_sizeB B)";
//     // atf_slack_post($msg, array('channel' => _atf_slack_chan_qa()));
    
//     $data['guid'] = $this->guid; 
//     $data['time'] = time();
//     $data['version'] = $version;
//     $data['revHash'] = $rev_hash;
//     $data['apk_size'] = $apk_sizeB;
//     atf_stats_send('apk_size', $data);

//     $data['guid'] = $this->guid; 
//     $data['time'] = time(); 
//     $data['version'] = $this->version; 
//     $data['revHash'] = $this->rev_hash; 
//     $data['branch'] = $this->branch; 
//     $data['event'] = 'apk_size';
//     $data['value'] = $apk_size;
//     atf_stats_send('event_value', $data);    
//   }
//   catch(Exception $e)
//   {
//     echo $e->getMessage() . "\n";
//   }
// }


// function trySendStats(ATFTask $task, $table, array $data)
// {
//   try
//   {
//     $data['guid'] = $this->guid; 
//     $data['time'] = time();
//     $data['deviceId'] = $task->device;
//     atf_stats_send($table, $data);
//   }
//   catch(Exception $e)
//   {
//     echo $e->getMessage() . "\n";
//   }
// }

// function trySendStatsEvent(ATFTask $task, $event, $value = 1)
// {
//   $this->trySendStats(
//     $task,
//     'event_value', 
//     array(
//       'event' => $event,
//       'value' => $value, 
//       'branch' => $this->branch, 
//       'version' => $this->version,
//       'revHash' => $this->rev_hash
//     )
//   );
// }

  // function tryShareToQAChannel($msg_slack_id, $error)
  // {
  //   if(!$this->share_with_qa_chan)
  //     return;

  //   //let's skip similar already shared errors 
  //   if($this->_calcSharedErrorSimilarity($error) > 80)
  //     return;
    
  //   $this->shared_qa_errors[] = $error;

  //   //let's share an exception message to goh-qa channel
  //   $resp = atf_slack_get_permalink($msg_slack_id);
  //   if(isset($resp['permalink']))
  //     atf_slack_post($resp['permalink'], array('channel' => _atf_slack_chan_qa()));
  // }

  function _calcSharedErrorSimilarity($error)
  {
    $max_perc = 0;
    foreach($this->shared_qa_errors as $shared)
    {
      $perc = 0;
      similar_text($shared, $error, $perc);
      if($perc > $max_perc)
        $max_perc = $perc;
    }
    return $max_perc;
  }
}


class ATFPlan
{
  const EXCEPTIONS_THRESHOLD = 5;
  const BOGUS_DEVICE_COUNT_THRESHOLD = 5;

  public $session;
  public $name;
  public $device_pool;
  public $tasks = array();
  public $slack_thread_ts;

  function __construct($name, IATFDevicePool $device_pool)
  {
    $this->name = $name;
    $this->device_pool = $device_pool;
  }

  function getTitle()
  {
    return "Testing plan '{$this->name}' devices:".sizeof($this->getDevices())."(n/a:".sizeof($this->session->ignored_devices).")";
  }

  // function createSlackThread()
  // {
  //   for($i=0;$i<5;++$i)
  //   {
  //     $resp = atf_slack_post($this->getTitle());
  //     if(isset($resp['ok']))
  //     {
  //       $this->slack_thread_ts = $resp['ts'];
  //       return;
  //     }
  //     sleep(1);
  //   }
  //   throw new Exception("Could not create Slack thread");
  // }

  // function updateSlackThread()
  // {
  //   $issues_summary = '';
  //   foreach($this->getProblemsHistogram() as $code => $count)
  //   {
  //     if($code == ATFTask::CODE_GONE)
  //       $issues_summary .= " gones:$count";
  //     else if($code == ATFTask::CODE_NSTART)
  //       $issues_summary .= " nstrts:$count";
  //     else if($code == ATFTask::CODE_STUCK)
  //       $issues_summary .= " stucks:$count";
  //     else if($code == ATFTask::CODE_HUNG)
  //       $issues_summary .= " hungs:$count";
  //     else if($code == ATFTask::CODE_EXCEPTION)
  //       $issues_summary .= " excepts:$count";
  //     else if($code == ATFTask::CODE_WARN)
  //       $issues_summary .= " warns:$count";
  //   }

  //   $running_count = 0;
  //   $over_count = 0;
  //   $total_progress = 0;

  //   foreach($this->tasks as $task)
  //   {
  //     $total_progress += $task->getProgress();
  //     if($task->isOver())
  //       $over_count++;
  //     if($task->device)
  //       $running_count++;
  //   }
  //   $progress_txt = ' tasks:' . ($running_count > 0 ? $running_count . '/' : '') . $over_count . '/' . sizeof($this->tasks) . ' (' . round($total_progress/sizeof($this->tasks)*100, 2) . '%)';

  //   atf_slack_update($this->slack_thread_ts,
  //     array(array(
  //       "text" => $this->getTitle() . $issues_summary . $progress_txt,
  //       "color" => $this->_getThreadColor(),
  //       "mrkdwn_in" => array("text")
  //     ))
  //   );
  // }

  function getProblemsHistogram()
  {
    $histogram = array();
    foreach($this->tasks as $task)
    {
      foreach($task->status_codes as $item)
      {
        list($code, $_) = $item;
        if(!ATFTask::isProblemCode($code))
          continue;

        if(!isset($histogram[$code]))
          $histogram[$code] = 0;
        ++$histogram[$code];
      }
    }
    return $histogram;
  }

  function hasStatusCode($code)
  {
    return $this->countStatusCode($code) > 0;
  }

  function countStatusCode($code)
  {
    $c = 0;
    foreach($this->tasks as $task)
      $c += $task->countStatusCode($code);
    return $c;
  }

  function _getThreadColor()
  {
    if(!$this->getDevices())
      return "#000000";

    $has_gones = $this->hasStatusCode(ATFTask::CODE_GONE);
    $has_hungs = $this->hasStatusCode(ATFTask::CODE_HUNG);
    $has_stucks = $this->hasStatusCode(ATFTask::CODE_STUCK);
    $has_nostarts = $this->hasStatusCode(ATFTask::CODE_NSTART);
    $has_excepts = $this->hasStatusCode(ATFTask::CODE_EXCEPTION);
    $has_warns = $this->hasStatusCode(ATFTask::CODE_WARN);

    if($has_gones || $has_hungs || $has_warns || $has_excepts || $has_stucks || $has_nostarts)
    {
      if(!$has_excepts)
        return $this->isOver() ? "warning" : "#FFCF9E";
      else
        return $this->isOver() ? "danger" : "#FFCCCB";
    }
    else if($this->isOver())
      return "good";
    else 
      return "#D3D3D3";
  }

  function _ignoreDevice($device, $reason)
  {
    $this->session->ignored_devices[] = $device;
    atf_log("Ignoring device *$device*: $reason");
    // atf_slack_post("Ignoring device *$device*: $reason", array('thread_ts' => $this->slack_thread_ts));
  }

  function getDevices()
  {
    $devices = array_diff($this->device_pool->get(), $this->session->ignored_devices);
    return $devices;
  }

  function addTask(ATFTask $task)
  {
    $this->tasks[] = $task;
  }

  function isOver()
  {
    if(!$this->getDevices())
    {
      return true;
    }
    if($this->countStatusCode(ATFTask::CODE_EXCEPTION) >= self::EXCEPTIONS_THRESHOLD)
    {
      return true;
    }
    foreach($this->tasks as $task)
    {
      if(!$task->isOver())
        return false;
    }
    return true;
  }

  function hasFatalProblems()
  {
    foreach($this->tasks as $task)
    {
      foreach($task->status_codes as $item)
      {
        if(ATFTask::isFatalCode($item[0]))
          return true;
      }
    }
    return false;
  }


  function _findUnassignedTask()
  {
    foreach($this->tasks as $task)
    {
      if(!$task->device && !$task->isOver())
        return $task;
    }
    return null;
  }

  function runAsync(ATFApkInstaller $install, $sleep_time, $hung_threshold, $gone_threshold, $stuck_threshold)
  {
    // $this->createSlackThread();

    $this->session->resetBogusDevices();

    $cs = array();

    foreach($this->tasks as $task)
      $cs[] = $this->_runTaskAsync($task, $install, $sleep_time, $hung_threshold, $gone_threshold, $stuck_threshold);

    return Amp\Promise\all($cs);
  }


  function _runTaskAsync(ATFTask $task, ATFApkInstaller $install, $sleep_time, $hung_threshold, $gone_threshold, $stuck_threshold)
  {
    return Amp\call(function() use($task, $install, $sleep_time, $hung_threshold, $gone_threshold, $stuck_threshold) {
      while(!$this->isOver() && !$task->isOver())
      {
        // $this->updateSlackThread();

        yield Amp\delay((int)$sleep_time*1000);

        if(!$task->device)
        {
          if(!yield $this->_findDeviceAndStartAsync($task, $install))
            continue;
        }

        $check_error = yield $this->_tryCheckExtStatusAsync($task, /*attempts*/3, /*timeout*/20);
        if($check_error !== null)
        {
          $this->_ignoreDevice($device, $check_error);
          if (!taskman_prop('EXT_BOT_ERROR_PAUSE'))
            $task->reschedule();
        }

        if(!$task->isOver())
          yield $this->_checkHealthAsync($task, $hung_threshold, $gone_threshold, $stuck_threshold);

        if($task->hasFatalProblem())
        {
          yield $this->_processFatalProblemAsync($task);
        }
        else if($task->isDone())
        {
          //let's reset hung stats for this device
          $this->session->resetBogusDeviceCount($task->device);
          //let's free the device
          $task->device = null;
        }
      }

      // $this->updateSlackThread();
    });
  }

  function _findDeviceAndStartAsync(ATFTask $task, ATFApkInstaller $install)
  {
    return Amp\call(function() use($task, $install) {

    $device = $this->_findFreeDevice();
    if(!$device)
      return false;

    //let's mark the device as occupied ASAP
    $task->device = $device;
    try
    {
      // atf_slack_post("Preparing *{$task->device}*", array('thread_ts' => $this->slack_thread_ts));

      yield $install->installAsync($device);
      yield atf_start_ext_cmd_on_device_async($device, $task->getCmd(), $task->getCmdArgs());
    }
    catch(Exception $e)
    {
      $this->_ignoreDevice($device, $e->getMessage());
      $task->reschedule();
      return false;
    }

    $task->start($this->slack_thread_ts);
    return true;

    });
  }

  function _processFatalProblemAsync(ATFTask $task)
  {
    return Amp\call(function() use($task) {

    // $this->_postScreenToSlack($task);

    $fatal_msg = "Fatal problem ({$task->last_fatal_problem} - ".ATFTask::code2string($task->last_fatal_problem)."), attempt:{$task->attempts} *{$task->device}*";

    atf_log("[FTL] $fatal_msg");

    // atf_slack_post($fatal_msg, array('thread_ts' => $task->slack_thread_ts));

    if($task->last_fatal_problem == ATFTask::CODE_GONE || 
       $task->last_fatal_problem == ATFTask::CODE_HUNG)  
    {
      $app_log = atf_get_logcat_unity($task->device, 300);
      $app_log = _atf_trim_start($app_log, 2000);
      // atf_slack_post("Last logs: ```$app_log``` *{$task->device}*", array('thread_ts' => $task->slack_thread_ts));
    }

    if (!taskman_prop('EXT_BOT_ERROR_PAUSE'))
      $task->reschedule();
      
    });
  } 


  function _findFreeDevice()
  {
    $available = array();
    
    foreach($this->getDevices() as $device)
    {
      $busy = false;
      foreach($this->tasks as $task)
      {
        if($task->device == $device)
        {
          $busy = true;
          break;
        }
      }
      if(!$busy)
       $available[] = $device;
    }
    
    if(!$available)
      return null;
    return $available[mt_rand(0, sizeof($available) - 1)];
  }

  function _tryCheckExtStatusAsync(ATFTask $task, $attempts, $timeout)
  {
    return Amp\call(function() use($task, $attempts, $timeout) {

    $last_error = null;

    for($i=0;$i<$attempts;++$i)
    {
      try
      {
        yield $this->_checkExtStatusAsync($task, $timeout);
        break;
      }
      catch(Exception $e)
      {
        $last_error = $e->getMessage();
        continue;
      }
    }

    return $last_error;

    });
  }

  function _checkExtStatusAsync(ATFTask $task, $timeout)
  {
    return Amp\call(function() use($task, $timeout) {

    $ext_status = yield atf_get_ext_status_async($task->device, $timeout);
    if(!is_array($ext_status))
      return;
    
    $new_items = $this->_getExtStatusItemsSince($task->last_ext_status_item_time, $ext_status);
    if(!$new_items)
      return;
    
    $task->last_ext_status_item_time = end($new_items)['time'];

    foreach($new_items as $item)
      yield $this->_analyzeExtStatusItemAsync($task, $item);
    
    });
  }


  function _getExtStatusItemsSince($time, array $ext_status)
  {
    $new_items = array();
    foreach($ext_status['entries'] as $item)
    {
      if($item['time'] > $time)
        $new_items[] = $item;
    }
    return $new_items;
  }


  function _checkHealthAsync(ATFTask $task, $hung_threshold, $gone_threshold, $stuck_threshold)
  {
    return Amp\call(function() use($task, $hung_threshold, $gone_threshold, $stuck_threshold) {

    $device_is_bogus = false;

    $not_alive_time = microtime(true) - $task->last_alive_check_time;
    $stuck_time = microtime(true) - $task->last_stuck_check_time;

    if($not_alive_time > $hung_threshold)
    {
      $task->addStatusCode(ATFTask::CODE_HUNG);
      // $this->session->trySendStatsEvent($task, 'hung');
      atf_log("[HNG] No activity for $hung_threshold seconds *{$task->device}*");
      $device_is_bogus = true;
    }
    else if($not_alive_time > $gone_threshold)
    {
      list($status, $_) = yield atf_host_exec_async("%{adb}% -s {$task->device} shell pidof %{package_id}%", DEPLOY_OPT_ERR_OK);
      if($status != 0)
      {
        if($task->last_alive_check_time === $task->reset_time)
        {
          $task->addStatusCode(ATFTask::CODE_NSTART);
          // $this->session->trySendStatsEvent($task, 'nstart');
          atf_log("[NFD] No app started after $hung_threshold seconds *{$task->device}*");
        }
        else
        {
          $task->addStatusCode(ATFTask::CODE_GONE);
          // $this->session->trySendStatsEvent($task, 'gone');
          atf_log("[GNE] App is gone after $hung_threshold seconds *{$task->device}*");
        }
        $device_is_bogus = true;
      }
      else 
        //let's tap the screen just in case app Activity is not in foreground
        yield atf_host_exec_async("%{adb}% -s {$task->device} shell input tap 360 930", DEPLOY_OPT_ERR_OK);
    }
    else if($stuck_time > $stuck_threshold)
    {
      $task->addStatusCode(ATFTask::CODE_STUCK);
      // $this->session->trySendStatsEvent($task, 'stuck');
      atf_log("[STK] Stuck for $stuck_threshold seconds *{$task->device}*");
    }

    if($device_is_bogus)
    {
      // $this->_reportErrorFromLogcatToSlack($task, 1000);
      $this->_incAndCheckBogusDevice($task->device);
    }

    });
  }

  function _incAndCheckBogusDevice($device)
  {
    if($this->session->incBogusDevice($device) >= self::BOGUS_DEVICE_COUNT_THRESHOLD)
    {
      $this->_ignoreDevice($device, "Too many N/As");
      return true;
    }
    return false;
  }

  static function _parseExtMessage($message)
  {
    $msg_code = null; 
    $msg_text = $message;
    //example: [DBG] this a debug message
    if(preg_match('~^(\[[^\]]+\])(.*)$~', $message, $matches))
    {
      $msg_code = $matches[1];
      $msg_text = $matches[2]; 
    }
    return array($msg_code, $msg_text);
  }

  static function _printToShellExtItem(ATFTask $task, array $item)
  {
    $shell_msg = _atf_trim($item['message'], 200);
    if(ATFTask::isProblemCode($item['error']))
      $shell_msg = "[PROBLEM] Code:{$item['error']}, $shell_msg";
    $shell_msg = "(".round($item['time'],1)."s) {$shell_msg} *{$task->device}*";

    atf_log($shell_msg);
  }

  // function _postToSlackExtStatusItem(ATFTask $task, array $item)
  // {
  //   $orig_msg = _atf_trim($item['message'], 3000);
  //   $slack_msg = $orig_msg;

  //   if($item['error'] == ATFTask::CODE_EXCEPTION)
  //     $slack_msg = "```$slack_msg```";

  //   $slack_msg = '('.round($item['time'],1).'s) '.$slack_msg.' *'.$task->device.'*';

  //   $resp = atf_slack_post($slack_msg, array('thread_ts' => $task->slack_thread_ts));

  //   if(isset($resp['ok']) && $item['error'] == ATFTask::CODE_EXCEPTION)
  //     $this->session->tryShareToQAChannel($resp['ts'], $orig_msg);
  // }

  function _analyzeExtStatusItemAsync(ATFTask $task, array $item)
  {
    return Amp\call(function() use($task, $item) {

    self::_printToShellExtItem($task, $item);

    $task->addStatusCode($item['error'], $item['message']);

    // if($item['error'] == ATFTask::CODE_EXCEPTION)
    //   $this->session->trySendStatsEvent($task, 'exception');

    list($msg_type, $msg_text) = self::_parseExtMessage($item['message']);

    //NOTE: alive system messages are sent periodically by ExtBot itself,
    //      we'd like to ignore them for stuck detection
    if($msg_type !== '[ALIVE]')
      $task->last_stuck_check_time = microtime(true);
    
    if($msg_type === '[DONE]')
    {
      $task->is_done = true;
      $this->_postToSlackExtStatusItem($task, $item);
    }

    //NOTE: in case of any message from the device we update the alive check timestamp
    $task->last_alive_check_time = microtime(true);

    yield $this->_processExtStatusMessageAsync($task, $item, $msg_type, $msg_text);

    });
  }

  function _processExtStatusMessageAsync($task, $item, $msg_type, $msg_text)
  {
    return Amp\call(function() use($task, $item, $msg_type, $msg_text) {

    if($msg_type === '[PRG]')
    {
      //let's reset hung stats since progress is going
      $this->session->resetBogusDeviceCount($task->device);
      $task->onProgress($msg_text);
    }
    // else if($msg_type === '[STAT]')
      // yield $this->session->trySendStatsFromJzonAsync($task, $msg_text);
    else if($msg_type === '[WRN]')
    {
      $task->addStatusCode(ATFTask::CODE_WARN);
      // $this->_postToSlackExtStatusItem($task, $item);
    }
    else if($msg_type === null)
    {
      // $this->_postToSlackExtStatusItem($task, $item);
    }

    });
  }

  // function _reportErrorFromLogcatToSlack(ATFTask $task, $limit)
  // {
  //   $errors_log = _atf_trim(atf_get_logcat_errors($task->device, $limit), 3000);
  //   if($errors_log)
  //     atf_slack_post("```$errors_log``` *{$task->device}*", array('thread_ts' => $task->slack_thread_ts));
  // }

  // function _postScreenToSlack(ATFTask $task)
  // {
  //   global $GAME_ROOT;

  //   $png_data = atf_screen($task->device);
  //   if($png_data)
  //   {
  //     ensure_write("$GAME_ROOT/build/atf/screen.png", $png_data);
  //     atf_slack_post_png("screen", "$GAME_ROOT/build/atf/screen.png", array('thread_ts' => $task->slack_thread_ts));
  //   }
  // }
}


class ATFTask 
{
  const CODE_NONE      = 0;
  const CODE_EXCEPTION = 1;
  const CODE_WARN      = 2;
  const CODE_NSTART    = 124;
  const CODE_STUCK     = 125;
  const CODE_GONE      = 126;
  const CODE_HUNG      = 127;
  const CODE_DONE      = 128;

  
  static function code2string($code)
  {
    switch($code)
    {
      case self::CODE_NONE:
        return "none";
      case self::CODE_EXCEPTION:
        return "exception";
      case self::CODE_WARN:
        return "warning";
      case self::CODE_STUCK:
        return "stuck";
      case self::CODE_NSTART:
        return "nstart";
      case self::CODE_GONE:
        return "gone";
      case self::CODE_HUNG:
        return "hung";
      case self::CODE_DONE:
        return "done";
    }
    return "???";
  }

  public $cmd;
  public $args;
  public $title;
  public $last_progress;
  public $last_done_arg_idx;
  public $slack_thread_ts;
  public $reset_time = 0;
  public $last_alive_check_time = 0;
  public $last_stuck_check_time = 0;
  public $last_ext_status_item_time = 0;
  public $last_ext_input_req_id = 0;
  public $status_codes = array();
  public $is_done;
  public $last_fatal_problem = 0;
  public $attempts = 1;
  public $device;

  function __construct($cmd, array $args, $title = '')
  {
    $this->cmd = $cmd;
    $this->args = $args;
    $this->title = $title;
  }

  function getCmd()
  {
    return $this->cmd;
  }

  function getCmdArgs()
  {
    return $this->args;
  }

  function getProgress()
  {
    return $this->last_progress;
  }

  static function isProblemCode($code)
  {
    return $code > self::CODE_NONE && $code < self::CODE_DONE;
  }

  static function isFatalCode($code)
  {
    return 
      $code == ATFTask::CODE_STUCK || 
      $code == ATFTask::CODE_NSTART || 
      $code == ATFTask::CODE_GONE || 
      $code == ATFTask::CODE_HUNG || 
      $code == ATFTask::CODE_EXCEPTION;
  }

  function hasFatalProblem()
  {
    return $this->last_fatal_problem != 0;
  }

  function hasStatusCode($code)
  {
    return $this->countStatusCode($code) > 0;
  }

  function countStatusCode($code)
  {
    $c = 0;
    foreach($this->status_codes as $item)
    {
      if($item[0] === $code)
        $c++;
    }
    return $c;
  }

  function isDone()
  {
    return $this->is_done;
  }

  function isOver()
  { 
    return $this->isDone() || $this->hasFatalProblem();
  }

  function addStatusCode($code, $msg = '')
  {
    $this->status_codes[] = array($code, $msg);

    if($code == ATFTask::CODE_DONE)
      $this->is_done = true;
    else if(self::isFatalCode($code))
      $this->last_fatal_problem = $code;
  }


  function onProgress($jzon)
  {
    try
    {
      $data = jzon_parse(trim(str_replace('\"', '"', $jzon)));
      if(isset($data['p']))
        $this->last_progress = floatval(str_replace(',', '.', str_replace('.', '', $data['p'])));
      if(isset($data['arg_idx']))
        $this->last_done_arg_idx = (int)$data['arg_idx'];
    }
    catch(Exception $e)
    {
      echo $e->getMessage() . "\n";
    }
  }

  function _resetCheckTimes()
  {
    $this->reset_time = microtime(true);
    $this->last_alive_check_time = $this->reset_time;
    $this->last_stuck_check_time = $this->reset_time;
    $this->last_ext_status_item_time = 0;
  }

  function reschedule()
  {
    ++$this->attempts;
    $this->device = null;
  }

  function start($slack_thread_ts)
  {
    $this->_resetCheckTimes();
    $this->slack_thread_ts = $slack_thread_ts;
  }
}

function atf_host_exec_async($cmd, $opts = 0, $timeout = 10) : Amp\Promise
{
  return Amp\call(function() use($cmd, $opts, $timeout) {

  $timeout_cmd = "/usr/local/bin/timeout -k 5s $timeout $cmd";

  $res = yield deploy_ssh_exec_async(atf_host(), $timeout_cmd, $opts);

  list($status, $lines) = current($res);
  //in case of timeout we set status to false explicitely
  if($status === 124)
    $status = false;

  return array($status, $lines);

  });
}

function atf_host_exec($cmd, $opts = 0, $timeout = 10)
{
  $timeout_cmd = "/usr/local/bin/timeout -k 5s $timeout $cmd";
 
  $res = deploy_ssh_exec(atf_host(), $timeout_cmd, $opts);

  list($status, $lines) = current($res);
  //in case of timeout we set status to false explicitely
  if($status === 124)
    $status = false;

  return array($status, $lines);
}

function atf_host_get_file($file_name)
{
  $file_data = current(deploy_get_file(atf_host(), $file_name));
  return $file_data;
}

function atf_host_put_file($local_path, $remote_path)
{
  deploy_scp_put_file(atf_host(), $local_path, $remote_path);
}

function atf_host_put_file_async($local_path, $remote_path)
{
  return deploy_scp_put_file_async(atf_host(), $local_path, $remote_path);
}

function _atf_slice(array $items, $max)
{
  $sliced = array();

  $offset = 0;
  while(true)
  {
    $tmp = array_slice($items, $offset, $max);
    if(!$tmp)
      break;
    $sliced[] = $tmp;
    $offset += $max;
  }

  return $sliced;
}

function _atf_get_remote_file_crc($path)
{
  list($status, $lines) = atf_host_exec("crc32 $path", DEPLOY_OPT_ERR_OK);
  if($status != 0)
    return '';
  return trim(implode("\n", $lines));
}

function _atf_get_local_file_crc($path)
{
  if(!is_file($path))
    return '';

  taskman_shell_ensure("crc32 $path", $out);
  return trim($out[0]);
}

function atf_device_put_file_async($device, $data_or_file, $device_path, $is_file = false)
{
  return Amp\call(function() use($device, $data_or_file, $device_path, $is_file) {

  if($device_path[0] !== "/")
    $device_path = atf_package_dir() . $device_path;

  $tmp_remote_file = uniqid(basename($device_path)."_");

  if(!$is_file)
  {
    $local_path = tempnam("/tmp", "atf_");
    ensure_write($local_path, $data_or_file);
  }
  else
    $local_path = $data_or_file;

  try
  {
    //1. let's copy local file to the ATF host as temp one
    yield atf_host_put_file_async($local_path, $tmp_remote_file);
  }
  finally
  {
    if(!$is_file)
      ensure_rm($local_path);
  }

  //2. let's push the temp file to the device
  $device_dir = dirname($device_path);
  yield atf_host_exec_async("%{adb}% -s $device shell mkdir -p $device_dir");
  yield atf_host_exec_async("%{adb}% -s $device push $tmp_remote_file $device_path", 0, 30);
  yield atf_host_exec_async("rm -rf $tmp_remote_file");

});
}

function atf_device_put_folder_async($device, $data, $device_path) 
{
  return Amp\call(function() use($device, $data, $device_path) {
    
    if($device_path[0] !== "/")
      $device_path = atf_package_dir() . $device_path;

    $device_dir = dirname($device_path);

    yield atf_host_exec_async("%{adb}% -s $device shell rm -rf $device_path");
    yield atf_host_exec_async("%{adb}% -s $device shell mkdir -p $device_dir");
    yield atf_host_exec_async("%{adb}% -s $device push $data $device_path", 0, 400);
  });
}

function atf_device_del_file_async($device, $device_path, $opts = 0)
{
  if($device_path[0] !== "/")
    $device_path = atf_package_dir() . $device_path;

  return atf_host_exec_async("%{adb}% -s $device shell rm -rf $device_path", $opts);
}

function atf_device_pull_file_async($device, $device_path, $timeout = 10, $throw_error_on_timeout = false)
{
  return Amp\call(function() use($device, $device_path, $throw_error_on_timeout, $timeout) {

  if($device_path[0] !== "/")
    $device_path = atf_package_dir() . $device_path;

  $tmp_remote_file = uniqid(basename($device_path)."_");

  list($status, $_) = yield atf_host_exec_async("%{adb}% -s $device pull $device_path $tmp_remote_file", DEPLOY_OPT_ERR_OK, $timeout);

  if($status !== 0)
  {
    if($status === false && $throw_error_on_timeout)
      throw new Exception("Could not pull from device '$device' file '$device_path' due to timeout");
    return null;
  }

  $file_data = atf_host_get_file($tmp_remote_file);
  yield atf_host_exec_async("rm -rf $tmp_remote_file");
  return $file_data;

  });
}

function atf_screen($device)
{
  $screen_file_name = uniqid("screen_");
  try
  {
    atf_host_exec("%{adb}% -s $device exec-out screencap -p > $screen_file_name");
  }
  catch(Exception $e)
  {
    return false;
  }
  $data = atf_host_get_file($screen_file_name);
  atf_host_exec("rm -rf $screen_file_name");
  return $data;
}

function atf_guid()
{
  return uniqid();
}

function _atf_start_watchdog($guid)
{
  $script = dirname(__FILE__) . '/../gamectl atf_watchdog ' . getmypid() . ' ' . $guid;
  exec("nohup $script > /dev/null &", $output, $ret);
}

function _atf_trim($txt, $max_len)
{
  return strlen($txt) > $max_len ? substr($txt, 0, $max_len) . "..." : $txt;
}

function _atf_trim_start($txt, $max_len)
{
  return strlen($txt) > $max_len ? "..." . substr($txt, strlen($txt) - $max_len) : $txt;
}


function atf_log($msg)
{
  echo date("Y-m-d H:i:s") . " " . $msg . "\n";
}

function atf_adb_reboot()
{
  atf_host_exec("killall adb");
  atf_host_exec("%{adb}% kill-server");
  atf_host_exec("%{adb}% start-server");
}

function atf_reboot_devices(array $devices)
{
  atf_log("Rebooting devices...");
  //let's reboot devices
  $ces = array();
  foreach($devices as $device)
  {
    $ces[] = Amp\call(function() use($device) {
      yield atf_host_exec_async("%{adb}% -s $device reboot", DEPLOY_OPT_ERR_OK);
      yield Amp\delay(40*1000);
    });
  }
  Amp\Promise\wait(Amp\Promise\all($ces));
}

function atf_get_devices($extended = false)
{
  $devices = array();

  list($_, $lines) = atf_host_exec("%{adb}% devices -l", DEPLOY_OPT_SILENT); 


  foreach($lines as $line)
  {
    if(preg_match('~^(\S+)\s+device(.*)~', $line, $m))
    {
      if($extended)
      {
        $devices[$m[1]] = array();
        $items = explode(" ", trim($m[2]));
        foreach($items as $item)
        {
          list($k, $v) = explode(":", $item);
          $devices[$m[1]][$k] = $v;
        }
      }
      else
        $devices[] = $m[1];
    }
  }

  return $devices;
}

function atf_get_ext_status_async($device, $timeout = 20)
{
  return Amp\call(function() use($device, $timeout) {
  
  $file_data = yield atf_device_pull_file_async($device, "ext_status.js", $timeout, /*throw on timeout*/true);
  if($file_data === null)
    return null;
  $file_data = preg_replace('/[[:cntrl:]]/', '', $file_data);
  return json_decode($file_data, true);

  });
}

function atf_update_ext_cmd_async($device, $main, array $args)
{
  $ext_cmd = array(
    "device_id" => $device,
    "main" => $main,
    "args" => $args,
  );

  return atf_device_put_file_async($device, json_encode($ext_cmd), 'ext_cmd.js');
}

function atf_package_dir()
{
  return "/storage/emulated/0/Android/data/".deploy_get(atf_host(), 'package_id')."/files/";
}

function atf_get_locat_errors_since($device, $since_stamp)//not using
{
  $time_spec = gmdate('m-d h:m:s.0', $since_stamp);
  list($_, $lines) = atf_host_exec("%{adb}% -s $device logcat -s -v UTC -d -t '$time_spec' AndroidRuntime:E Unity:E", DEPLOY_OPT_SILENT | DEPLOY_OPT_ERR_OK);
  _atf_filter_error_logs($lines);
  return implode("\n", $lines);
}

function atf_get_logcat_errors($device, $limit = 0)
{
  list($_, $lines) = atf_host_exec("%{adb}% -s $device logcat -s -d ".($limit > 0 ? "-t $limit" : "")." AndroidRuntime:E Unity:E", DEPLOY_OPT_SILENT | DEPLOY_OPT_ERR_OK);
  _atf_filter_error_logs($lines);
  return implode("\n", $lines);
}

function atf_get_logcat_unity($device, $limit = 0)
{
  list($_, $lines) = atf_host_exec("%{adb}% -s $device logcat -s -d Unity:'*'", DEPLOY_OPT_ERR_OK);
  if($limit > 0)
    array_splice($lines, 0, sizeof($lines) - $limit);
  return implode("\n", $lines);
}

function _atf_filter_error_logs(array &$lines)
{
  $exception_lines = 0;
  $filtered = array();
  for($i=0;$i<sizeof($lines);++$i)
  {
    $line = $lines[$i];
    if(strpos($line, 'E AndroidRuntime') !== false)
      $filtered[] = $line;
    else if(strpos($line, 'E Unity') !== false)
    {
      if(strpos($line, '[EXCEPTION]') !== false)
        $exception_lines = 15;

      if(--$exception_lines > 0)
        $filtered[] = $line;
    }
  }
  $lines = $filtered;
}

function atf_start_ext_cmd_on_device_async($device, $cmd_main, array $cmd_args)
{
  return Amp\call(function() use($device, $cmd_main, $cmd_args) {

  yield atf_device_del_file_async($device, 'ext_status.js', DEPLOY_OPT_ERR_OK);
  yield atf_host_exec_async("%{adb}% -s $device shell am force-stop %{package_id}%", DEPLOY_OPT_ERR_OK, 30);
  yield atf_update_ext_cmd_async($device, $cmd_main, $cmd_args);
  yield atf_host_exec_async("%{adb}% -s $device shell am start -n %{package_id}%/%{activity}%", 0, 30);

  });
}

function atf_host()
{
  static $inited = false;

  $node_name = 'atf';
  if($inited)
    return $node_name;

  deploy_declare_node($node_name, 
    array(
      'user' => taskman_prop("ATF_USER"),
      'dir' => '',
      'ssh_key_str' => taskman_prop("ATF_KEY"), 
      'hosts' => array(taskman_prop("ATF_HOST")),
      'adb' => taskman_prop("ATF_HOST_ADB"),
      'package_id' => taskman_prop("PACKAGE_ID"),
      'activity' => taskman_prop("LAUNCHER_ACTIVITY")
    )
  );

  $inited = true;

  return $node_name;
}


function _atf_opt(array &$args, $opt, $default, $conv_fn = null)
{
  foreach($args as $idx => $arg)
  {
    if(strpos($arg, $opt) === 0)
    {
      $arg = substr($arg, strlen($opt));
      unset($args[$idx]);
      return $conv_fn === null ? $arg : $conv_fn($arg);
    }
  }
  return $default;
}

function _atf_opt_check_no_trailing(array $args)
{
  foreach($args as $arg)
  {
    if(preg_match('~(--\w+)=.*~', $arg, $m))
      throw new Exception("Unknown option '{$m[1]}'");
  }
}