1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
#include <algorithm>
#include <climits>
#include <cstdio>
#include <cstdlib>
#include <string>
#include <vector>
#include "message.h"
#include "poszukiwania.h"
using namespace std;

long long S, M;

struct solved_partition {
  solved_partition(long long start_, long long end_, long long h_)
      : start(start_), end(end_), h(h_) { }

  bool operator < (const solved_partition& x) const {
    if (start == x.start) {
      if (end == x.end) {
        return h < x.h;
      }
      return end < x.end;
    }
    return start < x.start;
  }

  long long start, end, h;
};

#define H1 3425706325711LL

long long fast_pow(long long x, long long n) {
  long long v = 1LL;
  while (n > 0) {
    if (n & 1) {
      v = v * x;
    }
    
    x = x * x;
    n /= 2LL;
  }

  return v;
}

long long HashS(long long start, long long end, long long p) {
  long long h = 0;
  
  for (long long i = start; i <= end; i++) {
    h = (h * p) + SignalAt(i);
  }

  return h;
}

long long HashM(long long start, long long end, long long p) {
  long long h = 0;
  
  for (long long i = start; i <= end; i++) {
    h = (h * p) + SeqAt(i);
  }

  return h;
}

int DoTheCounting(long long start, long long end, long long h_s, long long h_m, long long p) {
  // Calculate p^S, required during the recalculation of the sequence
  // substring hash.
  long long p_pow_S = fast_pow(p, S);

  // Iterate over all requested offsets.
  int sum = 0;
  for (long long i = start; i <= end && i <= M - S + 1; i++) {
    if (h_m == h_s) {
      sum++;
    }
    
    // Adjust the hash to move the substring by one to the right.
    h_m *= p;
    h_m += SeqAt(i + S);
    h_m -= SeqAt(i) * p_pow_S;
  }

  return sum;
}

long long CalculatePartialHashM(const vector<solved_partition>& parts,
                                size_t idx) {
  long long count = 0;
  long long h = 0;
  long long local_start = parts[idx].start;
  while (count < S && idx < parts.size()) {
    long long expected_power = S - (parts[idx].start - local_start + 1);
    long long actual_power = (parts[idx].end - parts[idx].start + 1) - 1;
    h += parts[idx].h * fast_pow(H1, expected_power - actual_power);

    count += (parts[idx].end - parts[idx].start + 1);
    idx++;
  }

  return h;
}

int main() {
  // Get the S, M sequence lengths.
  S = SignalLength();
  M = SeqLength();

  if (M <= 1000000) {
    // If M (and therefore S) is reasonably small, just perform the calculations
    // locally in task 0, by calculating maximum pref-suff for S#M.
    if (MyNodeId() == 0) {
      vector<long long> w(1 + S + 1 + M);
      vector<long long> t(S + 1 + M);

      w[0] = -1;

      for (long long i = 0; i < S; i++) {
        t[i] = SignalAt(i + 1);
      }
      t[S] = INT_MAX;
      for (long long i = 0; i < M; i++) {
        t[S + 1 + i] = SeqAt(i + 1);
      }

      for (long long i = 0, j = -1; i < S + 1 + M; i++, j++, w[i] = j) {
        while ((j >= 0) && (t[i] != t[j])) {
          j = w[j];
        }
      }

      int sum = 0;
      for (long long i = 0; i < S + 1 + M; i++) {
        if (w[i + 1] == S) {
          sum++;
        }
      }

      printf("%d\n", sum);
    } else {
      // Do nothing in other tasks, just exit.
    }
  } else {
    long long items_per_task = M / (NumberOfNodes() - 1);
    if (S <= items_per_task) {
      // S is smaller than the workload each task has to handle in M. Therefore,
      // just assign the direct work to the machines, wait for the results,
      // aggregate them and print out.
      if (MyNodeId() == 0) {
        int sum = 0;
        int tasks_done = 0;

        while (tasks_done < NumberOfNodes() - 1) {
          int source = Receive(-1);
          sum += GetInt(source);
          tasks_done++;
        }

        printf("%d\n", sum);
      } else {
        // This is where actual work is done. Calculate the range of offsets
        // which needs to be covered by this task.
        long long start = 1 + (M * (MyNodeId() - 1)) / (NumberOfNodes() - 1);
        long long end;
        if (MyNodeId() != NumberOfNodes() - 1) {
          end = ((M * MyNodeId()) / (NumberOfNodes() - 1));
        } else {
          end = M - S;
        }

        // Calculate the overall hash for S.
        long long h_s = HashS(1, S, H1);
        
        // Calculate an initial hash for M.
        long long h_m = HashM(start, start + S - 1, H1);

        PutInt(0, DoTheCounting(start, end, h_s, h_m, H1));
        Send(0);
      }
    } else {
      // S is larger than a statistical workload of M for a single machine.
      // Therefore, we also have to split the hash calculation of S between
      // instances.
      if (MyNodeId() == 0) {
        // Partition S and assign tasks to compute hashes of substrings.
        for (int i = 1; i < NumberOfNodes(); i++) {
          long long start = 1 + (S * (i - 1)) / (NumberOfNodes() - 1);
          long long end;
          if (MyNodeId() != NumberOfNodes() - 1) {
            end = ((S * i) / (NumberOfNodes() - 1));
          } else {
            end = S;
          }

          PutInt(i, 1);  // 1: there's work incoming!
          PutInt(i, 0);  // 0: we're interested in a S hash.
          PutLL(i, start);  // starting index to hash.
          PutLL(i, end);  // end index to hash.
          Send(i);
        }

        // Receive the results and reassemble the hash.
        long long h_s = 0;
        int tasks_done = 0;
        while (tasks_done < NumberOfNodes() - 1) {
          int source = Receive(-1);
          long long start = GetLL(source);
          long long end = GetLL(source);
          long long h = GetLL(source);

          long long expected_power = S - start;
          long long actual_power = (end - start + 1) - 1;
          //printf("Received hash: [%lld, %lld]: %lld\n", start, end, h);
          //printf("Expected power: %lld, actual power: %lld\n", expected_power, actual_power);
          h_s += h * fast_pow(H1, expected_power - actual_power);
          tasks_done++;
        }

        //printf("Reassembled S hash: %lld\n", h_s);

        // Partition M and assign tasks to compute hashes of substrings.
        vector<long long> partitions;
        long long count = 0;
        long long S_left = S;

        while (count < M) {
          long long part = min(min(items_per_task, S_left), M - count);
          partitions.push_back(part);
          //printf("part: %lld\n", part);
          
          S_left -= part;
          if (S_left == 0) {
            S_left = S;
          }
          count += part;
        }

        //printf("Calculation tasks: %d, partitions: %lu\n", NumberOfNodes() - 1, partitions.size());

        // Assign the partitions to tasks for calculation.
        long long sum = 0;
        for (size_t i = 0; i < partitions.size(); i++) {
          int task_id = 1 + ((i * (NumberOfNodes() - 1)) / partitions.size());
          //printf("Assigning partition %lu to task %d\n", i, task_id);

          PutInt(task_id, 1);  // 1: there's work incoming!
          PutInt(task_id, 1);  // 1: we're interested in a M hash.
          PutLL(task_id, sum + 1);  // starting index to hash.
          PutLL(task_id, sum + partitions[i]); // end index to hash.
          Send(task_id);

          sum += partitions[i];
        }

        // Let all tasks know that it's the end of work.
        for (int i = 1; i < NumberOfNodes(); i++) {
          PutInt(i, 0);  // 0: no more hashing work, proceed to the next phase.
          Send(i);
        }

        // Receive all partial M hashes.
        size_t partitions_done = 0;
        vector<solved_partition> solved_partitions;
        while (partitions_done < partitions.size()) {
          int source = Receive(-1);
          long long start = GetLL(source);
          long long end = GetLL(source);
          long long h = GetLL(source);

          solved_partitions.emplace_back(start, end, h);

          partitions_done++;
        }

        // Sort the partitions.
        sort(solved_partitions.begin(), solved_partitions.end());

        //for (size_t i = 0; i < solved_partitions.size(); i++) {
        //  printf("Solved partition [%lld, %lld]: %lld\n",
        //      solved_partitions[i].start, solved_partitions[i].end, solved_partitions[i].h);
        //}

        // Count the partitions which actually need to be assigned.
        size_t assignable_partitions = 0;
        for (size_t i = 0; i < solved_partitions.size(); i++) {
          if (solved_partitions[i].start <= M - S + 1) {
            assignable_partitions++;
          } else {
            break;
          }
        }
        //printf("Assignable partitions: %d\n", assignable_partitions);

        // Assign partitions to tasks.
        for (size_t i = 0; i < assignable_partitions; i++) {
          int task_id = 1 + ((i * (NumberOfNodes() - 1)) / assignable_partitions);
          long long h_m = CalculatePartialHashM(solved_partitions, i);
          //printf("Assigning partition [%lld, %lld] to task %d for counting with hash %lld.\n",
          //    solved_partitions[i].start, solved_partitions[i].end, task_id, h_m);

          PutInt(task_id, 1);  // 1: there's work incoming!
          PutLL(task_id, solved_partitions[i].start);  // starting index for counting.
          PutLL(task_id, solved_partitions[i].end);  // ending index for couting.
          PutLL(task_id, h_s);  // overall S hash.
          PutLL(task_id, h_m);  // initial M hash.
          Send(task_id);
        }

        // Let all tasks know that now it's really the end of work.
        for (int i = 1; i < NumberOfNodes(); i++) {
          PutInt(i, 0);  // no more couting work.
          Send(i);
        }

        // Count results for all requests.
        int result = 0;
        partitions_done = 0;
        while (partitions_done < assignable_partitions) {
          int source = Receive(-1);
          result += GetInt(source);
          partitions_done++;
        }

        // Print out the result.
        printf("%d\n", result);
      } else {
        // Receive hashing requests from the master.
        int more_work = 0;
        do {
          Receive(0);
          more_work = GetInt(0);
          if (more_work) {
            int type = GetInt(0);
            long long start = GetLL(0);
            long long end = GetLL(0);
            //printf("[%d] Received [%d, %lld, %lld] request from master.\n",
            //    MyNodeId(), type, start, end);

            PutLL(0, start);
            PutLL(0, end);

            if (type == 0) {
              PutLL(0, HashS(start, end, H1));
            } else {
              PutLL(0, HashM(start, end, H1));
            }

            Send(0);
          }
        } while (more_work != 0);

        // Receive counting requests from the master.
        more_work = 0;
        do {
          Receive(0);
          more_work = GetInt(0);
          if (more_work) {
            long long start = GetLL(0);
            long long end = GetLL(0);
            long long h_s = GetLL(0);
            long long h_m = GetLL(0);
            //printf("[%d] Received [%lld, %lld, %lld, %lld] request from master.\n",
            //    MyNodeId(), start, end, h_s, h_m);

            PutInt(0, DoTheCounting(start, end, h_s, h_m, H1));
            Send(0);
          }
        } while (more_work != 0);
      }
    }
  }

  return 0;
}