parallelFilter [1029091]

< > BotCompany Repo | #1029091 // parallelFilter

JavaX fragment (include) [tags: use-pretranspiled]

// filters in parallel, returns elements in correct order
static <A> L<A> parallelFilter(Iterable<A> it, IPred<A> f) ctex {
  if (it == null) null;
  new L<Pair<A, Int>> out;
    
  int poolSize = coresToUse_fixed(), queueSize = 500;
  if (poolSize <= 1) ret filter(it, f);
  NotifyingBlockingThreadPoolExecutor e = new(poolSize, queueSize, 15, TimeUnit.SECONDS);

  try {
    int i = 0;
    for (A o : it) {
      ++i;
      int _i = i;
      e.execute(r { pcall {
        if (f.get(o)) synchronized(out) {
          out.add(pair(o, _i));
        }
      }});
    }
    e.shutdown();
    e.awaitTermination(1, TimeUnit.DAYS);
  } finally {
    e.shutdown();
  }

  ret firstOfPairs(sortBySecondOfPairs_inPlace(out));
}

Author comment

Began life as a copy of #1015446

Travelled to 7 computer(s): bhatertpkbcr, mqqgnosmbjvj, pyentgdyhuwx, pzhvpgtvlbxg, tvejysmllsmz, vouqrxazstgt, xrpafgyirdlv

Snippet ID:

#1029091

Snippet name:

parallelFilter

Eternal ID of this version:

#1029091/6

Text MD5:

ddd33b2825ea9e26931a1cb06692f3d5

Transpilation MD5:

e160cc85bce2c3ec52529e336ea529a6

Author:

stefan

Category:

javax

Type:

JavaX fragment (include)

Public (visible to everyone):

Yes

Archived (hidden from active list):

Created/modified:

2020-07-19 03:16:16

Source code size:

746 bytes / 28 lines

Pitched / IR pitched:

No / No

Views / Downloads:

428 / 555

Version history:

5 change(s)

Referenced in:

[show references]

1	// filters in parallel, returns elements in correct order
2	static <A> L<A> parallelFilter(Iterable<A> it, IPred<A> f) ctex {
3	if (it == null) null;
4	new L<Pair<A, Int>> out;
5
6	int poolSize = coresToUse_fixed(), queueSize = 500;
7	if (poolSize <= 1) ret filter(it, f);
8	NotifyingBlockingThreadPoolExecutor e = new(poolSize, queueSize, 15, TimeUnit.SECONDS);
9
10	try {
11	int i = 0;
12	for (A o : it) {
13	++i;
14	int _i = i;
15	e.execute(r { pcall {
16	if (f.get(o)) synchronized(out) {
17	out.add(pair(o, _i));
18	}
19	}});
20	}
21	e.shutdown();
22	e.awaitTermination(1, TimeUnit.DAYS);
23	} finally {
24	e.shutdown();
25	}
26
27	ret firstOfPairs(sortBySecondOfPairs_inPlace(out));
28	}