_binary From 1949044b9f70ac6b29e4de5512dd80b015889baa Mon Sep 17 00:00:00 2001
From: Gargi Sheguri <gargiegs@gmail.com>
Date: Sun, 23 Apr 2023 20:55:25 -0700
Subject: [PATCH] Sliding window addition

---
 src/controllers/CrawlController.php           |  12 +-
 src/controllers/FetchController.php           | 232 +++++++++++++++---
 src/controllers/components/CrawlComponent.php |   6 +-
 src/executables/Fetcher.php                   |  22 +-
 src/executables/QueueServer.php               |  92 ++++++-
 src/library/CrawlConstants.php                |   6 +
 src/library/CrawlDaemon.php                   |  44 +++-
 src/library/ReceiverMessage.php               |  67 +++++
 src/library/SenderMessage.php                 |  66 +++++
 src/models/CrawlModel.php                     |  12 +-
 src/models/MachineModel.php                   |  32 +++
 src/models/ParallelModel.php                  |   6 +-
 12 files changed, 546 insertions(+), 51 deletions(-)
 create mode 100644 src/library/ReceiverMessage.php
 create mode 100644 src/library/SenderMessage.php

diff --git a/src/controllers/CrawlController.php b/src/controllers/CrawlController.php
index 7fa562cf8..c1e0c94ef 100644
--- a/src/controllers/CrawlController.php
+++ b/src/controllers/CrawlController.php
@@ -58,7 +58,7 @@ class CrawlController extends Controller implements CrawlConstants
         "combinedCrawlInfo", "getInfoTimestamp", "getCrawlItems",
         "getCrawlList", "getCrawlSeedInfo", "sendStartCrawlMessage",
         "sendStopCrawlMessage", "setCrawlSeedInfo",
-        ];
+    ];
     /**
      * Checks that the request seems to be coming from a legitimate fetcher then
      * determines which activity the fetcher is requesting and calls that
@@ -307,9 +307,15 @@ class CrawlController extends Controller implements CrawlConstants
             $seed_info) = unserialize(L\webdecode($_REQUEST["arg"]));
         $seed_info['seed_sites']['url'] =
             L\partitionByHash(($seed_info['seed_sites']['url'] ?? ""),
-            null, $num, $i, C\NS_LIB . "UrlParser::getHost");
+                null, $num, $i, C\NS_LIB . "UrlParser::getHost");
+        if (isset($_REQUEST["active_fetchers"])) {
+            $active_fetchers = $_REQUEST["active_fetchers"];
+        }
+        else {
+            $active_fetchers = 0;
+        }
         $this->model("crawl")->sendStartCrawlMessage($crawl_params, $seed_info,
-            null, $num_fetchers);
+            null, $num_fetchers, $active_fetchers);
     }
     /**
      * Wrapper call to the source model method that deletes the news feed
diff --git a/src/controllers/FetchController.php b/src/controllers/FetchController.php
index 4d9957b56..080f6606d 100755
--- a/src/controllers/FetchController.php
+++ b/src/controllers/FetchController.php
@@ -64,6 +64,15 @@ class FetchController extends Controller implements CrawlConstants
      * alive)
      */
     const CRON_INTERVAL = 300;
+    /**
+     * Maximum amount of time (in secs) after which, if a sequence number 
+     * hasn't been acknowledged, it will be rescheduled
+     */
+    const TIMEOUT = 60;
+    /**
+     * Maximum number of times a fetch schedule can be rescheduled
+     */
+    const RESCHEDULE_TIMEOUT = 5;
     /**
      * Checks that the request seems to be coming from a legitimate fetcher then
      * determines which activity the fetcher is requesting and calls that
@@ -139,11 +148,61 @@ class FetchController extends Controller implements CrawlConstants
         } else {
             $crawl_time = 0;
         }
-        $schedule_filename = C\CRAWL_DIR . "/schedules/".
-            self::schedule_name . "$crawl_time.txt";
+        if (!file_exists(C\CRAWL_DIR . "/schedules/" .
+            self::sliding_window_base_name)) {
+            $window_info = ['SENDER' => [], 'RECEIVER' => [],
+                'NEXT_SEQ_NUM' => 1];
+        }
+        else {
+            $window_info_string = file_get_contents(C\CRAWL_DIR . "/schedules/"
+            . self::sliding_window_base_name);
+            $window_info = unserialize($window_info_string);
+            unset($window_info_string);
+            /*  iterate over sender window to check if any schedules have timed 
+                out and need to be rescheduled
+            */
+            if (!empty($window_info['SENDER'])) {
+                $time = time();
+                foreach ($window_info['SENDER'] as $i => $sender_object) {
+                    if ($sender_object->status == "s" && ($time -
+                            $sender_object->schedule_time) >= self::TIMEOUT) {
+                        if ($sender_object->reschedule_count >
+                            self::RESCHEDULE_TIMEOUT) {
+                            $schedule_filename = C\CRAWL_DIR . "/schedules/" .
+                                self::schedule_name . $crawl_time . "_" .
+                                $sender_object->seq_num . ".txt";
+                            if (file_exists($schedule_filename)) {
+                                unlink($schedule_filename);
+                            }
+                            unset($window_info['SENDER'][$i]);
+                        }
+                        else {
+                            $seq_num = $sender_object->seq_num;
+                            $sender_object->schedule_time = $time;
+                            $sender_object->reschedule_count++;
+                            break;
+                        }
+                    }
+                }
+            }
+        }
+        //if none of the schedules have timed out, schedule the next seq number
+        if (!isset($seq_num)) {
+            $seq_num = $window_info['NEXT_SEQ_NUM'];
+        }
+        $schedule_filename = C\CRAWL_DIR . "/schedules/" .
+            self::schedule_name . $crawl_time . "_" . $seq_num . ".txt";
         if (file_exists($schedule_filename)) {
+            if ($seq_num == $window_info['NEXT_SEQ_NUM']) {
+                $window_info['SENDER'][] =
+                    new L\SenderMessage($seq_num, time(), "s", 0);
+                $window_info['NEXT_SEQ_NUM']++;
+            }
+            $fh = fopen(C\CRAWL_DIR . "/schedules/" .
+                self::sliding_window_base_name, "w");
+            fwrite($fh, serialize($window_info));
+            fclose($fh);
             $data['MESSAGE'] = file_get_contents($schedule_filename);
-            unlink($schedule_filename);
         } else {
             /*  check if scheduler part of queue server went down
                 and needs to be restarted with current crawl time.
@@ -321,7 +380,9 @@ class FetchController extends Controller implements CrawlConstants
                     we only set crawl time. Other data such as allowed sites
                     should come from index.
                 */
-                $this->model("crawl")->sendStartCrawlMessage($crawl_params);
+                $this->model("crawl")->sendStartCrawlMessage($crawl_params,
+                    active_fetchers_count: $this->model("machine")
+                        ->getActiveFetchersCount());
             }
         }
     }
@@ -335,7 +396,7 @@ class FetchController extends Controller implements CrawlConstants
         $info_flag = false;
         $logging = "";
         $necessary_fields = ['byte_counts', 'current_part', 'hash_data',
-            'hash_part', 'num_parts', 'part'];
+            'hash_part', 'num_parts', 'part', 'seq_num', 'diff_origin'];
         $part_flag = true;
         $missing = "";
         $channel = $this->getChannel();
@@ -349,38 +410,114 @@ class FetchController extends Controller implements CrawlConstants
             $this->checkRestart($this->clean(
                 $_REQUEST['crawl_type'], 'string'));
         }
+        if (isset($_REQUEST['crawl_time'])) {;
+            $crawl_time = substr($this->clean($_REQUEST['crawl_time'], 'int'),
+                0, C\TIMESTAMP_LEN);
+        } else {
+            $crawl_time = 0;
+        }
         if ($part_flag &&
             L\crawlHash($_REQUEST['part']) == $_REQUEST['hash_part']) {
             $upload = false;
+            if (!file_exists(C\CRAWL_DIR . "/temp")) {
+                mkdir(C\CRAWL_DIR . "/temp");
+                L\setWorldPermissionsRecursive(C\CRAWL_DIR . "/temp/");
+            }
+            $filename = C\CRAWL_DIR . "/temp/" . $_REQUEST['hash_data'];
+            file_put_contents($filename, $_REQUEST['part'], FILE_APPEND);
+            L\setWorldPermissions($filename);
             if (intval($_REQUEST['num_parts']) > 1) {
                 $info_flag = true;
-                if (!file_exists(C\CRAWL_DIR . "/temp")) {
-                    mkdir(C\CRAWL_DIR . "/temp");
-                    L\setWorldPermissionsRecursive(C\CRAWL_DIR . "/temp/");
-                }
-                $filename = C\CRAWL_DIR . "/temp/" . $_REQUEST['hash_data'];
-                file_put_contents($filename, $_REQUEST['part'], FILE_APPEND);
-                L\setWorldPermissions($filename);
                 if ($_REQUEST['num_parts'] == $_REQUEST['current_part']) {
                     $upload = true;
                 }
             } else if (intval($_REQUEST['num_parts']) == 1) {
                 $info_flag = true;
                 $upload = true;
-                $filename = "";
             }
             if ($upload) {
+                $diff_origin = filter_var($_REQUEST['diff_origin'],
+                    FILTER_VALIDATE_BOOLEAN);
+                $seq_num = intval($_REQUEST['seq_num']);
+                $byte_counts = [];
+                if (!empty($_REQUEST['byte_counts'])) {
+                    $byte_counts = unserialize(
+                        L\webdecode($_REQUEST['byte_counts']));
+                }
+                $schedule_time = $byte_counts["SCHEDULE_TIME"];
+                $incoming_receiver_message = new L\ReceiverMessage($seq_num, 
+                time(), $diff_origin, $filename);
+                $window_info_string =
+                    file_get_contents(C\CRAWL_DIR . "/schedules/" .
+                        self::sliding_window_base_name);
+                $window_info = unserialize($window_info_string);
+                unset($window_info_string);
+                /*  acknowledge the received schedule in $sender if it 
+                originated from the current queue server
+                */
+                if (!$diff_origin) {
+                    if (!$diff_origin && !empty($window_info['SENDER']) && 
+                        $seq_num < $window_info['SENDER'][0]->seq_num) {
+                        $this->handleUploadedData(
+                            $incoming_receiver_message->filename);
+                        $inserted = true;
+                    }
+                    else {
+                        foreach ($window_info['SENDER'] as $sender_obj) {
+                            if ($seq_num == $sender_obj->seq_num) {
+                                $sender_obj->status = "a";
+                                //unlink the corresponding schedule.txt file
+                                $schedule_filename = C\CRAWL_DIR . "/schedules/"
+                                    . self::schedule_name . "$crawl_time" . "_" 
+                                    . $seq_num . ".txt";
+                                if (file_exists($schedule_filename)) {
+                                    unlink($schedule_filename);
+                                }
+                                break;
+                            }
+                        }
+                    }
+                }
+
+                /*  only receive data either scheduled by the current queue 
+                    server, or by a different queue server if it fits in the 
+                    current window
+                */
+                if (!isset($inserted) && (!$diff_origin ||
+                        (!empty($window_info['SENDER']) && $schedule_time >
+                            $window_info['SENDER'][0]->schedule_time))) {
+                    if (!empty($window_info['RECEIVER'])) {
+                        $receiver_len = count($window_info['RECEIVER']);
+                        for ($i = 0; $i < $receiver_len; $i++) {
+                            if (($diff_origin && $schedule_time <
+                                $window_info['RECEIVER'][$i]->schedule_time) || 
+                                (!$diff_origin &&
+                                $window_info['RECEIVER'][$i]->diff_origin &&
+                                $schedule_time <
+                                $window_info['RECEIVER'][$i]->schedule_time) || 
+                                (!$diff_origin &&
+                                !$window_info['RECEIVER'][$i]->diff_origin && 
+                                $seq_num < 
+                                $window_info['RECEIVER'][$i]->seq_num)) {
+                                array_splice($window_info['RECEIVER'], $i,
+                                    0, [$incoming_receiver_message]);
+                                $inserted = true;
+                                break;
+                            }
+                        }
+                    }
+                    if (!isset($inserted)) {
+                        $window_info['RECEIVER'][] = $incoming_receiver_message;
+                    }
+                }
                 $upload_old = false;
                 if (file_exists($this->crawl_status_file_name)) {
                     $crawl_status = unserialize(file_get_contents(
                         $this->crawl_status_file_name));
-                    if (!empty($crawl_status['REPEAT_TIME']) &&
-                        !empty($_REQUEST['byte_counts'])) {
-                        $byte_counts = unserialize(
-                            L\webdecode($_REQUEST['byte_counts']));
-                        if (!empty($byte_counts["SCHEDULE_TIME"]) &&
-                            $crawl_status['REPEAT_TIME'] >
-                            $byte_counts["SCHEDULE_TIME"]) {
+                    if (!empty($crawl_status['REPEAT_TIME']) && 
+                        isset($byte_counts)) {
+                        if (isset($schedule_time) &&
+                            $crawl_status['REPEAT_TIME'] > $schedule_time) {
                             $upload_old = true;
                         }
                     }
@@ -388,7 +525,50 @@ class FetchController extends Controller implements CrawlConstants
                 if ($upload_old) {
                     $logging = "DoubleIndexArchive has switched index!";
                 } else {
-                    $logging = $this->handleUploadedData($filename);
+                    $logging = "... Data received successfully\n";
+                    if (!empty($window_info['RECEIVER']) &&
+                        !$window_info['RECEIVER'][0]->diff_origin) {
+                        while (!empty($window_info['SENDER']) &&
+                            !empty($window_info['RECEIVER'])
+                            && $window_info['SENDER'][0]->seq_num ==
+                            $window_info['RECEIVER'][0]->seq_num) {
+                            array_shift($window_info['SENDER']);
+                            $upload_data = array_shift(
+                                $window_info['RECEIVER']);
+                            $msg = $this->handleUploadedData(
+                                $upload_data->filename);
+                            L\crawlLog($msg . " for schedule sequence number " .
+                                $upload_data->seq_num);
+                            /*  write any diff_origin schedule data to inverted 
+                                index files
+                            */
+                            while (!empty($window_info['RECEIVER']) &&
+                                $window_info['RECEIVER'][0]->diff_origin) {
+                                $upload_data = array_shift(
+                                    $window_info['RECEIVER']);
+                                $this->handleUploadedData(
+                                    $upload_data->filename);
+                            }
+                        }
+                        /*  write any remaining diff_origin schedule data to 
+                            inverted index files after sender window is empty
+                        */
+                        if (empty($window_info['SENDER'])) {
+                            while (!empty($window_info['RECEIVER'])) {
+                                $upload_data = array_shift(
+                                    $window_info['RECEIVER']);
+                                $this->handleUploadedData(
+                                    $upload_data->filename);
+                            }
+                        }
+                    }
+                }
+                $sliding_window_file = C\CRAWL_DIR . "/schedules/" .
+                    self::sliding_window_base_name;
+                if (file_exists($sliding_window_file)) {
+                    $fh = fopen($sliding_window_file, "w");
+                    fwrite($fh, serialize($window_info));
+                    fclose($fh);
                 }
             } else {
                 $logging = "...".(
@@ -491,13 +671,9 @@ class FetchController extends Controller implements CrawlConstants
      */
     public function handleUploadedData($filename = "")
     {
-        if ($filename == "") {
-            $uploaded = $_REQUEST['part'];
-        } else {
-            $uploaded = file_get_contents($filename);
-            unlink($filename);
-        }
-        $logging = "... Data upload complete\n";
+        $uploaded = file_get_contents($filename);
+        unlink($filename);
+        $logging = "... Data upload complete";
         $address = strtr(L\remoteAddress(), ["." => "-", ":" => "_"]);
         $time = time();
         $day = floor($time/C\ONE_DAY);
diff --git a/src/controllers/components/CrawlComponent.php b/src/controllers/components/CrawlComponent.php
index fbc3e85a8..57af5000d 100644
--- a/src/controllers/components/CrawlComponent.php
+++ b/src/controllers/components/CrawlComponent.php
@@ -234,7 +234,8 @@ class CrawlComponent extends Component implements CrawlConstants
                         serialize($crawl_params));
                     chmod($filename, 0777);
                     if ($crawl_model->sendStartCrawlMessage($crawl_params,
-                        null, $machine_urls)) {
+                        null, $machine_urls, active_fetchers_count:
+                        $parent->model("machine")->getActiveFetchersCount())) {
                         return $parent->redirectWithMessage(
                             tl('crawl_component_resume_crawl'),
                             $request_fields);
@@ -696,7 +697,8 @@ class CrawlComponent extends Component implements CrawlConstants
         $parent->web_site->filePutContents($filename, serialize($crawl_params));
         chmod($filename, 0777);
         if (!$crawl_model->sendStartCrawlMessage($crawl_params,
-            $seed_info, $machine_urls)) {
+            $seed_info, $machine_urls, active_fetchers_count:
+            $parent->model("machine")->getActiveFetchersCount())) {
             $parent->redirectWithMessage(
                 tl('crawl_component_start_fail'),
                 $request_fields);
diff --git a/src/executables/Fetcher.php b/src/executables/Fetcher.php
index 09fbca663..c41e5ed24 100755
--- a/src/executables/Fetcher.php
+++ b/src/executables/Fetcher.php
@@ -439,6 +439,12 @@ class Fetcher implements CrawlConstants
      * @var string
      */
     public $debug;
+    /**
+     * Holds the sequence number of the current schedule received from 
+     * queue server
+     * @var int
+     */
+    public $seq_num;
     /**
      * Before receiving any data from a queue server's web app this is
      * the default assumed post_max_size in bytes
@@ -531,6 +537,7 @@ class Fetcher implements CrawlConstants
         $this->summarizer_option = self::BASIC_SUMMARIZER;
         $this->debug = "";
         $this->num_multi_curl = C\NUM_MULTI_CURL_PAGES;
+        $this->seq_num = 1;
     }
     /**
      * Return the fetcher's copy of a page processor for the given
@@ -1305,6 +1312,10 @@ class Fetcher implements CrawlConstants
         }
         $info = unserialize($decode_token);
         $this->setCrawlParamsFromArray($info);
+        if (isset($info[self::SEQUENCE_NUM])) {
+            L\crawlLog("Received schedule number: " . 
+            $info[self::SEQUENCE_NUM]);
+        }
         if (isset($info[self::SITES])) {
             $tok = strtok("\n"); //skip meta info
             $this->to_crawl = [];
@@ -1470,6 +1481,8 @@ class Fetcher implements CrawlConstants
         $cs = $this->current_server;
         do {
             if ($at_least_current_server ||
+                (isset($this->found_sites[self::SEEN_URLS][$cs]) &&
+                    count($this->found_sites[self::SEEN_URLS][$cs]) > 0) ||
                 (isset($this->found_sites[self::TO_CRAWL][$cs]) &&
                 count($this->found_sites[self::TO_CRAWL][$cs]) > 0)) {
                 $this->updateScheduler($cs, $at_least_current_server);
@@ -1648,6 +1661,9 @@ class Fetcher implements CrawlConstants
         if (isset($info[self::MAX_LINKS_TO_EXTRACT])) {
             $this->max_links_to_extract = $info[self::MAX_LINKS_TO_EXTRACT];
         }
+        if (isset($info[self::SEQUENCE_NUM])) {
+            $this->seq_num = $info[self::SEQUENCE_NUM];
+        }
     }
     /**
      * Prepare an array of up to NUM_MULTI_CURL_PAGES' worth of sites to be
@@ -2663,7 +2679,9 @@ class Fetcher implements CrawlConstants
             'crawl_time' => $this->crawl_time, 'machine_uri' => C\WEB_URI,
             'robot_instance' => $prefix . C\ROBOT_INSTANCE, 'data' => '',
             'check_crawl_time' => $this->check_crawl_time,
-            'crawl_type' => $this->crawl_type];
+            'crawl_type' => $this->crawl_type,
+            'diff_origin' => ($server != $this->current_server),
+            'seq_num' => $this->seq_num];
         //handle robots.txt data
         if (isset($this->found_sites[self::ROBOT_TXT]) && $send_robots) {
             $data = L\webencode(
@@ -2717,8 +2735,8 @@ class Fetcher implements CrawlConstants
             L\crawlLog("..." . $bytes_schedule . " bytes of schedule data");
             $byte_counts["TOTAL"] += $bytes_schedule;
             $byte_counts["SCHEDULE"] = $bytes_schedule;
-            $byte_counts["SCHEDULE_TIME"] = $this->schedule_time;
         }
+        $byte_counts["SCHEDULE_TIME"] = $this->schedule_time;
         unset($schedule_data);
         if ($seen_cnt > 0 || !empty(
             $this->found_sites[self::LINK_SEEN_URLS][$server])) {
diff --git a/src/executables/QueueServer.php b/src/executables/QueueServer.php
index bc4e3dd18..25e414be2 100644
--- a/src/executables/QueueServer.php
+++ b/src/executables/QueueServer.php
@@ -314,14 +314,24 @@ class QueueServer implements CrawlConstants
      * @var string
      */
     public $debug;
+    /**
+     * Indicates the maximum number of schedules generated at a time
+     * (if these many schedules are created already, queue server pauses on
+     * schedule creation until one or more of them have been successfully
+     * downloaded)
+     * The window size is set to be the total number of currently active 
+     * fetchers
+     * @var int
+     */
+    public $window_size;
     /**
      * Holds the int value of the previous partition in index
      * @var int
      */
     public $last_next_partition_to_add;
     /**
-     * Keeps track of the time needed for the dictionary updater to add the current
-     * partition contents to index
+     * Keeps track of the time needed for the dictionary updater to add the 
+     * current partition contents to index
      * @var int
      */
     public $start_dictionary_time;
@@ -398,6 +408,7 @@ class QueueServer implements CrawlConstants
         $this->server_name = "IndexerAndScheduler";
         $this->process_name = "0-QueueServer";
         $this->debug = "";
+        $this->window_size = 1;
     }
     /**
      * This is the function that should be called to get the queue server
@@ -850,11 +861,14 @@ class QueueServer implements CrawlConstants
                 }
                 L\measureCall($this, "processToCrawlUrls");
                 L\measureCall($this, "processReceivedRobotTxtUrls");
-                if (!file_exists(
-                    C\CRAWL_DIR . "/schedules/" . self::schedule_name.
-                    $this->crawl_time . ".txt")) {
-                        L\measureCall($this, "produceFetchBatch");
+                if(file_exists(C\CRAWL_DIR . "/schedules/"
+                        . self::next_seq_num_file) && count(glob(C\CRAWL_DIR
+                        . "/schedules/" . self::schedule_name. $this->crawl_time
+                        . "_*")) < $this->window_size) {
+                    L\measureCall($this, "produceFetchBatch");
                 }
+                else L\crawlLog("Sender window full: Next schedule will 
+                be produced when current schedules are acknowledged");
                 break;
             case self::ARCHIVE_CRAWL:
                 if ($this->isAIndexer()) {
@@ -1028,6 +1042,9 @@ class QueueServer implements CrawlConstants
             } else {
                 return $old_info;
             }
+            if (isset($info[self::ACTIVE_FETCHERS_COUNT])) {
+                $this->window_size = $info[self::ACTIVE_FETCHERS_COUNT];
+            }
             switch ($info[self::STATUS])
             {
                 case "NEW_CRAWL":
@@ -1042,8 +1059,15 @@ class QueueServer implements CrawlConstants
                         }
                         $this->stopCrawl();
                     }
+                    //initialize files for sliding window scheduling
+                    $current_crawl_ts_file = C\CRAWL_DIR .
+                        "/schedules/" . self::current_crawl_timestamp_file;
+                    file_put_contents($current_crawl_ts_file, 
+                        $info[self::CRAWL_TIME]);
+                    file_put_contents(C\CRAWL_DIR . "/schedules/" .
+                        self::next_seq_num_file, 1);
                     $this->startCrawl($info);
-                    $info[self::STATUS] == self::CONTINUE_STATE;
+                    $info[self::STATUS] = self::CONTINUE_STATE;
                     if (!$is_scheduler) {
                         L\crawlLog("Starting new crawl. Timestamp:" .
                             $this->crawl_time);
@@ -1101,8 +1125,45 @@ class QueueServer implements CrawlConstants
                             }
                             $this->stopCrawl();
                         }
+                        /* shift all cache/WindowData for the crawl back to 
+                            /schedules
+                        */
+                        $current_crawl_ts_file = C\CRAWL_DIR .
+                            "/schedules/" . self::current_crawl_timestamp_file;
+                        file_put_contents($current_crawl_ts_file,
+                            $info[self::CRAWL_TIME]);
+                        if (file_exists(C\CRAWL_DIR . "/cache/" .
+                            self::window_data_base_name . 
+                            $info[self::CRAWL_TIME] . "/" . 
+                            self::sliding_window_base_name)) {
+                            rename(C\CRAWL_DIR . "/cache/" .
+                                self::window_data_base_name . 
+                                $info[self::CRAWL_TIME] . "/" . 
+                                self::sliding_window_base_name,
+                                C\CRAWL_DIR . "/schedules/" .
+                                self::sliding_window_base_name);
+                        }
+                        if (file_exists(C\CRAWL_DIR . "/cache/" .
+                            self::window_data_base_name . 
+                            $info[self::CRAWL_TIME] . "/" . 
+                            self::next_seq_num_file)) {
+                            rename(C\CRAWL_DIR . "/cache/" .
+                                self::window_data_base_name .
+                                $info[self::CRAWL_TIME] . "/" . 
+                                self::next_seq_num_file,
+                                C\CRAWL_DIR . "/schedules/" .
+                                self::next_seq_num_file);
+                        }
+                        $schedule_files = glob(C\CRAWL_DIR . "/cache/" .
+                            self::window_data_base_name . 
+                            $info[self::CRAWL_TIME] . "/" . 
+                            self::schedule_name . "*");
+                        foreach ($schedule_files as $schedule_file) {
+                            rename($schedule_file, C\CRAWL_DIR . "/schedules/" .
+                                basename($schedule_file));
+                        }
                         $this->startCrawl($info);
-                        $info[self::STATUS] == self::CONTINUE_STATE;
+                        $info[self::STATUS] = self::CONTINUE_STATE;
                         if (!$is_scheduler) {
                             L\crawlLog("Resuming crawl");
                             $this->writeAdminMessage("RESUME_CRAWL");
@@ -1139,6 +1200,7 @@ class QueueServer implements CrawlConstants
      */
     public function stopCrawl()
     {
+        CrawlDaemon::saveWindowData();
         if ($this->isAScheduler()) {
             $this->crawl_queue = null;
         }
@@ -2270,6 +2332,8 @@ class QueueServer implements CrawlConstants
         $sites[self::MAX_LINKS_TO_EXTRACT] = $this->max_links_to_extract;
         $sites[self::POST_MAX_SIZE] = L\metricToInt(ini_get("post_max_size"));
         $sites[self::SITES] = [];
+        $sites[self::SEQUENCE_NUM] = intval(file_get_contents(C\CRAWL_DIR .
+            "/schedules/" . self::next_seq_num_file));
         return base64_encode(serialize($sites)) . "\n";
     }
     /**
@@ -2561,9 +2625,12 @@ class QueueServer implements CrawlConstants
                 }
             }
             ksort($sites);
+            $next_sequence_num = intval(file_get_contents(C\CRAWL_DIR .
+                "/schedules/" . self::next_seq_num_file));
             //write schedule to disk
             $fh = fopen(C\CRAWL_DIR . "/schedules/" .
-                self::schedule_name . $this->crawl_time . ".txt", "wb");
+                self::schedule_name . $this->crawl_time . "_" .
+                $next_sequence_num . ".txt", "wb");
             fwrite($fh, $first_line);
             $num_sites = count($sites);
             $k = 0;
@@ -2638,9 +2705,14 @@ class QueueServer implements CrawlConstants
                 L\changeInMicrotime($new_time));
             L\crawlLog("FB Scheduler: End Produce Fetch Batch Memory usage: " .
                 memory_get_usage() );
-            L\crawlLog("FB Scheduler: Created fetch batch of size $num_sites." .
+            L\crawlLog("FB Scheduler: Created fetch batch " .
+                $next_sequence_num . " of size $num_sites." .
                 "...Total Time to create batch: ".
                 L\changeInMicrotime($start_time));
+            $next_sequence_num++;
+            file_put_contents(C\CRAWL_DIR . "/schedules/" .
+                self::next_seq_num_file, $next_sequence_num);
+            unset($next_sequence_num);
         }
     }
     /**
diff --git a/src/library/CrawlConstants.php b/src/library/CrawlConstants.php
index b1a78d3b0..51b4d91f8 100755
--- a/src/library/CrawlConstants.php
+++ b/src/library/CrawlConstants.php
@@ -71,9 +71,13 @@ interface CrawlConstants
     const fetch_closed_name = "FetchClosed";
     const data_base_name = "At";
     const schedule_start_name = "StartCrawlSchedule.txt";
+    const window_data_base_name = "WindowData";
+    const sliding_window_base_name = "SlidingWindow.txt";
     const robot_table_name = "robot_table.txt";
     const mirror_table_name = "mirror_table.txt";
     const local_ip_cache_file = "local_ip_cache.txt";
+    const next_seq_num_file = "next_seq_num.txt";
+    const current_crawl_timestamp_file = "current_crawl_timestamp.txt";
     /** used for word iterator direction  */
     const ASCENDING = 1;
     const DESCENDING = -1;
@@ -260,4 +264,6 @@ interface CrawlConstants
     const PROCESSOR = 'eo';
     const FALLBACK_PROCESSOR = 'ep';
     const SCRAPER_INFO = 'eq';
+    const SEQUENCE_NUM = 'er';
+    const ACTIVE_FETCHERS_COUNT = 'es';
 }
diff --git a/src/library/CrawlDaemon.php b/src/library/CrawlDaemon.php
index da7408e5c..6e8d0fef9 100644
--- a/src/library/CrawlDaemon.php
+++ b/src/library/CrawlDaemon.php
@@ -398,10 +398,14 @@ class CrawlDaemon implements CrawlConstants
             crawlLog("$name_string does not appear to running...");
         }
         if ($exit) {
-            if ($name == 'index') {
+            if ($name == 'QueueServer') {
+                CrawlDaemon::saveWindowData();
+            }
+            elseif ($name == 'index') {
                 $global_stop_file = C\WORK_DIRECTORY .
                     "/data/global_stop.txt";
                 file_put_contents($global_stop_file, time());
+                CrawlDaemon::saveWindowData();
                 chmod($global_stop_file, 0777);
                 exit();
             }
@@ -491,4 +495,42 @@ class CrawlDaemon implements CrawlConstants
         }
         return $active_daemons;
     }
+
+    /**
+     * Save current window data files (fetch schedules produced that haven't 
+     * been acknowledged yet, next_seq_num to be scheduled by queue server, and 
+     * sliding window information from the fetch controller) to 
+     * cache/WindowData dir
+     */
+    public static function saveWindowData()
+    {
+        $current_crawl_ts_file = C\CRAWL_DIR .
+            "/schedules/" . self::current_crawl_timestamp_file;
+        if (file_exists($current_crawl_ts_file)) {
+            $current_crawl_ts = file_get_contents($current_crawl_ts_file);
+            $window_data_dirname = C\CRAWL_DIR . "/cache/" . 
+            self::window_data_base_name . $current_crawl_ts;
+            if (!is_dir($window_data_dirname)) {
+                mkdir($window_data_dirname);
+            }
+            if (file_exists(C\CRAWL_DIR . "/schedules/" .
+                self::sliding_window_base_name)) {
+                rename(C\CRAWL_DIR . "/schedules/" . 
+                self::sliding_window_base_name, $window_data_dirname . "/" . 
+                self::sliding_window_base_name);
+            }
+            if (file_exists(C\CRAWL_DIR . "/schedules/" .
+                self::next_seq_num_file)) {
+                rename(C\CRAWL_DIR . "/schedules/" . self::next_seq_num_file,
+                    $window_data_dirname . "/" . self::next_seq_num_file);
+            }
+            $schedule_files = glob(C\CRAWL_DIR . "/schedules/" .
+                self::schedule_name . $current_crawl_ts . "_*");
+            foreach ($schedule_files as $schedule_file) {
+                rename($schedule_file,
+                    $window_data_dirname . "/" . basename($schedule_file));
+            }
+            unlink($current_crawl_ts_file);
+        }
+    }
 }
diff --git a/src/library/ReceiverMessage.php b/src/library/ReceiverMessage.php
new file mode 100644
index 000000000..403e5cdf3
--- /dev/null
+++ b/src/library/ReceiverMessage.php
@@ -0,0 +1,67 @@
+<?php
+/**
+ *  SeekQuarry/Yioop --
+ *  Open Source Pure PHP Search Engine, Crawler, and Indexer
+ *
+ *  Copyright (C) 2009 - 2023  Chris Pollett chris@pollett.org
+ *
+ *  LICENSE:
+ *
+ *  This program is free software: you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation, either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program.  If not, see .
+ *
+ *  END LICENSE
+ *
+ * @package seek_quarry
+ * @subpackage executables
+ * @license http://www.gnu.org/licenses/ GPL3
+ * @link http://www.seekquarry.com/
+ * @copyright 2009 - 2023
+ * @filesource
+ */
+namespace seekquarry\yioop\library;
+
+class ReceiverMessage
+{
+    /**
+     * Sequence number of received webpage schedule
+     * @var int
+     */
+    public $seq_num;
+    /**
+     * Timestamp of when object was scheduled (to sort receiver objects in 
+     * current window by time of scheduling)
+     * @var int
+     */
+    public $schedule_time;
+    /**
+     * Set if the scheduling queue server is different from the current 
+     * (destination)
+     * @var bool
+     */
+    public $diff_origin;
+    /**
+     * Filename for corresponding posted data in /temp dir
+     * @var string
+     */
+    public $filename;
+
+    public function __construct($seq_num, $schedule_time, $diff_origin, 
+    $filename)
+    {
+        $this->seq_num = $seq_num;
+        $this->schedule_time = $schedule_time;
+        $this->diff_origin = $diff_origin;
+        $this->filename = $filename;
+    }
+}
\ No newline at end of file
diff --git a/src/library/SenderMessage.php b/src/library/SenderMessage.php
new file mode 100644
index 000000000..43d3920c8
--- /dev/null
+++ b/src/library/SenderMessage.php
@@ -0,0 +1,66 @@
+<?php
+/**
+ *  SeekQuarry/Yioop --
+ *  Open Source Pure PHP Search Engine, Crawler, and Indexer
+ *
+ *  Copyright (C) 2009 - 2023  Chris Pollett chris@pollett.org
+ *
+ *  LICENSE:
+ *
+ *  This program is free software: you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation, either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program.  If not, see .
+ *
+ *  END LICENSE
+ *
+ * @package seek_quarry
+ * @subpackage executables
+ * @license http://www.gnu.org/licenses/ GPL3
+ * @link http://www.seekquarry.com/
+ * @copyright 2009 - 2023
+ * @filesource
+ */
+namespace seekquarry\yioop\library;
+
+class SenderMessage
+{
+    /**
+     * Sequence number of schedule
+     * @var int
+     */
+    public $seq_num;
+    /**
+     * Scheduling timestamp
+     * @var int
+     */
+    public $schedule_time;
+    /**
+     * Keeps track of schedule's current status, i.e. either scheduled (s) or
+     * acknowledged (a)
+     * @var string
+     */
+    public $status;
+    /**
+     * Keeps track of the number of times a fetch schedule has been rescheduled
+     * @var int
+     */
+    public $reschedule_count;
+
+    public function __construct($seq_num, $schedule_time, $status, 
+    $reschedule_count)
+    {
+        $this->seq_num = $seq_num;
+        $this->schedule_time = $schedule_time;
+        $this->status = $status;
+        $this->reschedule_count = $reschedule_count;
+    }
+}
\ No newline at end of file
diff --git a/src/models/CrawlModel.php b/src/models/CrawlModel.php
index 2491f7dd8..7e0366032 100755
--- a/src/models/CrawlModel.php
+++ b/src/models/CrawlModel.php
@@ -926,6 +926,7 @@ EOT;
         $delete_dirs = [
             C\CRAWL_DIR . '/cache/' . self::double_index_base_name . $timestamp,
             C\CRAWL_DIR . '/cache/' . self::index_data_base_name . $timestamp,
+            C\CRAWL_DIR . '/cache/' . self::window_data_base_name . $timestamp,
             C\CRAWL_DIR . '/schedules/' . self::index_data_base_name .
                 $timestamp,
             C\CRAWL_DIR . '/schedules/' . self::schedule_data_base_name .
@@ -988,9 +989,11 @@ EOT;
      * @param int $num_fetchers number of fetchers on machine to start.
      *      This parameter and $channel are used to start the daemons
      *      running on the machines if they aren't already running
+     * @param int $active_fetchers_count number of active fetchers running 
+     * across all yioop instances currently
      */
     public function sendStartCrawlMessage($crawl_params, $seed_info = null,
-        $machine_urls = null, $num_fetchers = 0)
+        $machine_urls = null, $num_fetchers = 0, $active_fetchers_count = 0)
     {
         if ($machine_urls != null && !$this->isSingleLocalhost($machine_urls)) {
             $params = [$crawl_params, $seed_info];
@@ -1003,7 +1006,8 @@ EOT;
                 self::SLEEP_START => $sleep_start,
                 self::SLEEP_DURATION => $sleep_duration]));
             $this->execMachines("sendStartCrawlMessage",
-                $machine_urls, serialize($params), 0, true);
+                $machine_urls, serialize($params), 0, true,
+                $active_fetchers_count);
             return true;
         }
         $channel = (empty($crawl_params[self::CHANNEL])) ? 0 :
@@ -1015,7 +1019,9 @@ EOT;
                 return false;
             }
         }
-        $info_string = serialize($crawl_params);
+        $info = $crawl_params;
+        $info[self::ACTIVE_FETCHERS_COUNT] = $active_fetchers_count;
+        $info_string = serialize($info);
         file_put_contents(
             C\CRAWL_DIR . "/schedules/$channel-QueueServerMessages.txt",
             $info_string);
diff --git a/src/models/MachineModel.php b/src/models/MachineModel.php
index 79bf2fd6d..adc4c3ec0 100644
--- a/src/models/MachineModel.php
+++ b/src/models/MachineModel.php
@@ -532,4 +532,36 @@ class MachineModel extends Model
         }
         return substr($job, 0, -3);
     }
+    /**
+     * Returns the total number of active fetchers across all machines
+     * @return int number of fetchers currently turned 'on'
+     */
+    public function getActiveFetchersCount()
+    {
+        // get the total number of fetchers that have been turned 'on'
+        $machine_list = $this->getMachineList();
+        $machine_statuses = $this->getMachineStatuses($machine_list);
+        $active_fetchers_count = 0;
+        foreach ($machine_statuses as $machine => $value) {
+            if (array_key_exists('STATUSES', $value) &&
+                array_key_exists('Fetcher', $value['STATUSES'])) {
+                if (is_array($value['STATUSES']['Fetcher'])) {
+                    foreach ($value['STATUSES']['Fetcher'] as $status) {
+                        if ($status) {
+                            $active_fetchers_count++;
+                        }
+                    }
+                }
+                else {
+                    foreach ($value['STATUSES']['Fetcher'] as 
+                        $fetcher => $status) {
+                        if ($status) {
+                            $active_fetchers_count++;
+                        }
+                    }
+                }
+            }
+        }
+        return $active_fetchers_count;
+    }
 }
diff --git a/src/models/ParallelModel.php b/src/models/ParallelModel.php
index 96e269904..759799493 100755
--- a/src/models/ParallelModel.php
+++ b/src/models/ParallelModel.php
@@ -520,10 +520,12 @@ class ParallelModel extends Model
      * @param int $num_machines the integer to be used in calculating partition
      * @param bool $send_specs whether to send the queue_server, num fetcher
      *      info for given machine
+     * @param int $active_fetchers_count number of active fetchers running 
+     * across all yioop instances currently
      * @return array a list of outputs from each machine that was called.
      */
     public function execMachines($command, $machine_urls, $arg = null,
-        $num_machines = 0, $send_specs = false)
+        $num_machines = 0, $send_specs = false, $active_fetchers_count = 0)
     {
         if (empty($machine_urls) || !is_array($machine_urls)) {
             return [];
@@ -534,7 +536,7 @@ class ParallelModel extends Model
         $time = time();
         $session = md5($time . C\AUTH_KEY);
         $query = "c=crawl&a=$command&time=$time&session=$session" .
-            "&num=$num_machines";
+            "&num=$num_machines&active_fetchers=$active_fetchers_count";
         if ($arg !== null) {
             $arg = L\webencode($arg);
             $query .= "&arg=$arg";
-- 
2.39.1

