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
|
import time
import gc
import os
import epics
import pvnames
pvlist = pvnames.updating_pvlist
def show_memory():
gc.collect()
if os.name == 'nt':
return 'Windows memory usage?? pid=%i' % os.getpid()
f = open("/proc/%i/statm" % os.getpid())
mem = f.readline().split()
f.close()
return 'Memory: VmSize = %i kB / VmRss = %i kB' %( int(mem[0])*4 , int(mem[1])*4)
N_new = 0
def get_callback(pvname=None,value=None,**kw):
global N_new
N_new = N_new + 1
def monitor_events(t = 60.0):
print( 'Processing PV requests:')
t0 = time.time()
endtime = t0 + t
global N_new
nnotify = 10
while time.time() < endtime:
epics.ca.pend_event(0.01)
if N_new >= nnotify:
print( "Saw %i changes in %.3f seconds: %s" % (N_new, time.time()-t0, show_memory()))
N_new = 0
t0 = time.time()
def run(t=10.0):
pvs = [epics.PV(i, callback=get_callback) for i in pvlist]
epics.ca.pend_io(1.0)
for p in pvs: p.get()
print( 'Monitoring %i PVs' % len(pvs))
monitor_events(t=t)
print( 'Destroying PVs: ')
for i in pvs:
i.disconnect()
print( epics.ca._cache.keys())
epics.ca.show_cache()
epics.ca.poll(0.01, 10.0)
time.sleep(1.0)
for i in range(4):
print( "==run # ", i+1)
run(t=15)
print( 'memory leak test complete.')
|
