FilesExpand file tree

 main

/

test_concurrent_cache.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

95 lines (80 loc) · 2.89 KB

 main

/

test_concurrent_cache.py

Copy path

Top

File metadata and controls

• Code
• Blame

95 lines (80 loc) · 2.89 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

import sys

import os.path

import unittest

import multiprocessing

import semmle

from tests import test_utils

from semmle.util import makedirs

ITERATIONS = 100

CONCURRENCY = 20

class ConcurrentCacheTest(test_utils.ExtractorTest):

'''

Test the cache under heavy concurrent load.

'''

def __init__(self, name):

super(ConcurrentCacheTest, self).__init__(name)

self.cachedir = os.path.abspath(os.path.join(self.here, "cache"))

def setUp(self):

super(ConcurrentCacheTest, self).setUp()

makedirs(self.cachedir)

self.cache = semmle.cache.Cache(self.cachedir)

def tearDown(self):

super(ConcurrentCacheTest, self).tearDown()

def _concurrent_read_and_write(self):

readers = []

writers = []

queue = multiprocessing.Queue(CONCURRENCY+1)

for i in range(CONCURRENCY):

readers.append(multiprocessing.Process(target=read_func, args=(self.cache, queue)))

writers.append(multiprocessing.Process(target=write_func, args=(self.cache, ITERATIONS//4)))

for read, write in zip(readers, writers):

read.start()

write.start()

for proc in writers:

proc.join()

for proc in readers:

proc.join()

successes = [ queue.get(False) for i in range(CONCURRENCY) ]

self.assertNotIn(None, successes)

# We expect a fairly low success rate here

# But want to assert that at least one read succeeded.

self.assertGreater(sum(successes), 0)

def _concurrent_read_ok(self):

readers = []

queue = multiprocessing.Queue(CONCURRENCY+1)

for i in range(CONCURRENCY):

readers.append(multiprocessing.Process(target=read_func, args=(self.cache, queue)))

for proc in readers:

proc.start()

for proc in readers:

proc.join()

successes = [ queue.get(False) for i in range(CONCURRENCY) ]

self.assertNotIn(None, successes)

self.assertEqual(sum(successes), 2*CONCURRENCY*ITERATIONS)

def test(self):

#Must run this first as it populates the cache

self._concurrent_read_and_write()

#Then this tests that the cache is correctly populated.

self._concurrent_read_ok()

def key(i):

return "key%d" % i

def value(i):

return ("value%d\n" % i).encode("utf-8")*10000

def read_func(cache, queue):

successes = 0

for i in range(ITERATIONS):

val = cache.get(key(i))

if val is not None:

successes += 1

assert val == value(i)

for i in range(ITERATIONS):

val = cache.get(key(i))

if val is not None:

successes += 1

assert val == value(i)

queue.put(successes)

def write_func(cache, offset):

for i in range(offset, ITERATIONS):

cache.set(key(i), value(i))

for i in range(offset-1, -1, -1):

cache.set(key(i), value(i))