Loop is eating my memory.

[robert]

First of all I would like to thank everybody that is working with this excellent program.

I've written a plugin that uses OpenOPC http://openopc.sourceforge.net/ to communicate with a KNX OPC-server.

The problem is that the while-loop is eating up my memory very fast.

Code: Select all

ii = 0
while ii < i:
	getvalue = opc.read(knxitem[ii], timeout=30000)
	KNXstatus = getvalue[0]
	if KNXstatus:
		if readstatus[ii] == 0:
		eg.TriggerEvent("TRUE",payload = knxdescription[ii],prefix = knxgroup[ii])
		readstatus[ii] = 1
	else:
		if readstatus[ii] == 1:	
		eg.TriggerEvent("FALSE",payload = knxdescription[ii],prefix = knxgroup[ii])
		readstatus[ii] = 0
	ii = ii + 1

Is there a way to clear the memory when the loop is finished or what am I doing wrong?

Thanks in advance
Robert

[Pako]

I can not try your script, because I do not have an OPC server.
What is the value of variable i ?
Maybe it's too big number and your loop is close to never ending while loop.
It is known that a similar loop eats a lot of memory.

Pako

[krambriw]

I think it is the while loop itself, you are not giving back the resources to the system, the cpu is very busy
Try to add a time.sleep in the loop. And I think you should move the code for reading the OPC to a separate thread

I have tried to access my OPC servers and it works fine, thanks for the hint about OpenOPC! Have worked a lot with OPC servers & clients from various suppliers, this is very interesting


Best regrads, Walter

[robert]

Hi again,

'i' is the number of variables read from the OPC server, in this case 54.
So that should not be a problem.

After the loop is finished I do have a sleep.time for 1 second.
Right now I have to restart EventGhost every 12 hour because the memory of the computer have run out (2Gb).

I have tried to run this loop in a different thread but then I got the following error message.

Code: Select all

14:08:28   Unhandled exception in thread started by 
14:08:28   Traceback (most recent call last):
14:08:28     File "C:\Program Files\EventGhost\plugins\OPCServer\__init__.py", line 142, in ReadLoop
14:08:28       getValue = opc.read(knxItem[ii], timeout=30000)
14:08:28     File "C:\Program Files\EventGhost\lib26\site-packages\OpenOPC.py", line 610, in read
14:08:28       return list(results)[0]
14:08:28     File "C:\Program Files\EventGhost\lib26\site-packages\OpenOPC.py", line 405, in iread
14:08:28       raise OPCError, error_msg
14:08:28   OpenOPC.OPCError: AddGroup: The application called an interface that was marshalled for a different thread.

Any idea what that's all about.

Best regards
Robert

[krambriw]

Try to post your complete code, otherwise it is difficult for us to guess. If you tend to keep your plugin private, expect difficulties with support.

Code: Select all

getValue = opc.read(knxItem[ii], timeout=30000)

If you have defined the opc interface in the plugin code, you may try to declare it as self.opc....

Then the thread should be able to access it via something like

Code: Select all

getValue = self.opc.read(knxItem[ii], timeout=30000)

[robert]

Ok, I haven't got the new thread code to work the way that I want, obviously I'm doing something wrong.

But here is the original code, I had to make a lot of changes to be able to show it to you (it looked like crap...).

Code: Select all

import eg
import OpenOPC
import time
from threading import Event, Thread

eg.RegisterPlugin(
    name = "OPCServer",
    author = "Robert Berggren",
    version = "0.0.1",
    kind = "other",
    description = "This plugin reads values from a NETxKNX OPCServer."
)

################################################

class ReadOPCServer(eg.PluginBase):

    def __init__(self):
        print "OPCServer is inited."

#################################################

    def __start__(self, myString):
        print "OPCServer is started" + myString
        self.stopThreadEvent = Event()
        thread = Thread(target=self.ThreadLoop, args=(self.stopThreadEvent, ))
        thread.start()

#################################################

    def __stop__(self):
        print "OPCServer is stopped."
	self.stopThreadEvent.set()

#################################################

    def __close__(self):
        print "OPCServer is closed."
	self.stopThreadEvent.set()

#################################################

    def Configure(self, myString=""):
        panel = eg.ConfigPanel()
        textControl = wx.TextCtrl(panel, -1, myString)
        panel.sizer.Add(textControl, 1, wx.EXPAND)
        while panel.Affirmed():
            panel.SetResult(textControl.GetValue())
            
#################################################
            
    def ThreadLoop(self, stopThreadEvent):

        opc = OpenOPC.client()
        opc.connect('NETxKNX.OPC.Server.3.5')

        line = []
        knxGroup = []
        knxItem = []
        knxDescription = []
        currentStatus = []
        
        f1 = open('C:\Program Files\EventGhost\plugins\OPCServer\KNX_Items.txt')

#Sortera gruppadresser och Beskrivningar       
        numberOfItems = 0
        for readline in f1:
            line.append(readline)
            tmp = line[numberOfItems]
            x = tmp.rfind('/n')
            knxDescription.append(tmp[31:x])
            knxGroup.append(tmp[22:30])
            tmp = tmp[1:9] + tmp[10:20] + tmp[21:30]
            knxItem.append(tmp[:27])
            currentStatus.append(numberOfItems)
            numberOfItems = numberOfItems + 1
        f1.close()
        
# Läs startvärden
        ii = 0
        while ii < numberOfItems:    	
            getStatus = opc.read(knxItem[ii], timeout=30000)
            knxStatus = getStatus[0]
            if knxStatus:
		currentStatus[ii] = 1
            else:
	   	currentStatus[ii] = 0
            print knxGroup[ii] + ' ; ' + knxDescription[ii] + ' ; ' + str(knxStatus)
            ii = ii + 1

##################################################
            
	while not stopThreadEvent.isSet():      

#Läs och skriv ut STATUS från NETxKNX
            ii = 0
            while ii < numberOfItems:
                getStatus = opc.read(knxItem[ii], timeout=30000)
                knxStatus = getStatus[0]
                if knxStatus:
                    if currentStatus[ii] == 0:
                        eg.TriggerEvent("TRUE",payload = knxDescription[ii],prefix = knxGroup[ii])
                        currentStatus[ii] = 1
                else:
                    if currentStatus[ii] == 1:	
                        eg.TriggerEvent("FALSE",payload = knxDescription[ii],prefix = knxGroup[ii])
                        currentStatus[ii] = 0
                ii = ii + 1
                
            stopThreadEvent.wait(1.0)
            
        opc.close()
#################################################

[krambriw]

Hi,

I quickly also wrote an OPC plugin and also mine is consuming memory (the used memory is counting up for every opc.read I do)

I have tried various things to make the release of memory but so far nothing has helped. Only way to avoid memory consumption found so far is to set the EG option "Limit memory consumption while minimized to XY Mbytes" and to run EG minimized

Obviously EG is able to release memory during run time but I have not yet found how to activate this from code. If we know this, we could add this to the plugin

Best regards, Walter

[krambriw]

Hi again,

Attached is my simple OPC plugin implementation.

OPC.zip

(29.06 KiB) Downloaded 236 times

Hopefully it will also find your installed KNX OPC server. I do not believe there is a memory leak because memory is released and given back to the OS. This can be studied by setting a low value of memory usage in the options and then minimize EG.

I have found a way to use this memory release feature and to limit the memory usage also in non minimized mode, using a little trick (maybe some have an opinion on this, maybe some would think this should be a part of EG option settings itself)

The trick I use is the following

In the code I have the following import:

Code: Select all

import eg
import OpenOPC
import time
from eg.WinApi.Dynamic import (
    OpenProcess,
    PROCESS_SET_QUOTA,
    SetProcessWorkingSetSize,
    FormatError,
)

In the init of the plugin, I have added:

Code: Select all

    def __init__(self):
        self.hHandle = OpenProcess(PROCESS_SET_QUOTA, 0, eg.processId)

The code for the thread looks like this:

Code: Select all

    def ThreadWorker(self, stopThreadEvent):
        eg.config.limitMemorySize = self.limitMemorySize
        while not stopThreadEvent.isSet():
            stopThreadEvent.wait(self.iDelay)
            if not self.bOpcObjectCreated:
                self.findOpcServer()
            for i in range (0,len(self.tags)):
                tag = str(self.tags[i])
                #print tag
                tag_items = self.items[i]
                #print tag_items
                for j in range (0,len(tag_items)):
                    value, quality, time = self.opc.read(tag_items[j])
                    print tag_items[j], "Value: ", value
            if self.q > 100 or self.q == 0:
                SetProcessWorkingSetSize(
                    self.hHandle,
                    3670016,
                    self.limitMemorySize * 1048576
                )
                self.q = 1
            self.q += 1
            
        self.opc.close()
        self.bOpcObjectCreated = False
        print self.text.infoThreadStopped

Specifically note the following:

Code: Select all

        eg.config.limitMemorySize = self.limitMemorySize

and

Code: Select all

                SetProcessWorkingSetSize(
                    self.hHandle,
                    3670016,
                    self.limitMemorySize * 1048576
                )

The setting dialog for the OPC server looks typically like this with a drop-down to select the OPC server you like to connect to. The setting for maximum memory usage will be used in all modes (minimized, maximized, windowed)

[robert]

Hi again,

thank you for trying to help me with this.

I have tried to set the maximum amount of memory for EventGhost to 50Mb and that limits only the eventghost.exe process.
If I look on "PF usage" under "Performance" in the Task Manager the memory still increases very fast.

I will try to implement your solution into my plugin as soon as I got the time and see if that helps.

Best regards
Robert

[krambriw]

OPC_2011.05.27.zip

(15.84 KiB) Downloaded 241 times

I have made a change to the memory consumption part. Use this version instead

[krambriw]

I have made some bugs regarding handling of threads when you add the plugin...looking into this

The idea with memory limitations are still the same

[krambriw]

Now hopfully, this will work better...

OPC_2011.05.27a.zip

(19.89 KiB) Downloaded 233 times

[robert]

Now I have had the time to try your solution, but I can't make it work.
It behaves the same as I have described before.
This code only limits the memory for eventghost.exe if you look at processes in the Task Manager.
If you look at the memory usage in the Performance page in the task Manager the memory still increases.

Is it the same for you?

Best regards
Robert

[krambriw]

Is it the same for you?

Yes, unfortunately it seems. Memory is increasing, you can see how EG is trying to release memory but almost right away the consumption is up at the same level again. I do not know what could be wrong. Could be a problem related to the OpenOPC also (Gateway Service). I think the increase in memory consumption (how fast it grows) is depending on how quickly you poll the interface

Best regards, Walter

[krambriw]

I think I found a way to overcome the problem (don't really like it but what to do?).

I think I have noticed that the memory is handed back to the os when you close the connection to the opc server (please verify if this is also your opinion)

I have modified the code in my thread so that it makes this disconnection & re-connection automatically. See this example as a draft. It would be possible to do more sophisticated, i.e. at a certain time/day or when the memory consumption has reached a certain level. In the code below, I just do it after a number of read operations

Check and see if this could work, I have not seen any drawbacks yet

Best regards, Walter

Code: Select all

    def ThreadWorker(self, stopThreadEvent):
        self.hHandle = OpenProcess(PROCESS_SET_QUOTA, 0, eg.processId)
        while not stopThreadEvent.isSet():
            stopThreadEvent.wait(self.iDelay)
            if not self.bOpcObjectCreated:
                self.findOpcServer()
            for i in range (0,len(self.tags)):
                tag = self.tags[i]
                tag_items = self.items[i]
                for j in range (0,len(tag_items)):
                    value, quality, time = self.opc.read(tag_items[j])
                    tmp = self.previousStatus[i]
                    if value != tmp[j]:
                        eg.TriggerEvent(tag_items[j],payload = value)
                        tmp[j] = value
                self.previousStatus[i] = tmp
            if self.q > 50:
                self.opc.close()
                stopThreadEvent.wait(0.2)
                print "OPC re-connecting"
                self.opc.connect(self.OPC_name)
                self.q = 1
            self.q += 1
            
        self.opc.close()
        self.bOpcObjectCreated = False
        print self.text.infoThreadStopped