ProtoTree x->right ==x Caused infinite loop

[honglei]

Call Stack：

 	Norm.dll!ProtoTree::Bit(const char * key=0x000001db527ca670, unsigned int keysize=96, unsigned int index=64, ProtoTree::Endian keyEndian=ENDIAN_BIG) Line 232	C++
>	Norm.dll!ProtoTree::Remove(ProtoTree::Item & item={...}) Line 459	C++
 	[Inline Frame] Norm.dll!ProtoTreeTemplate<ProtoSortedTree::Item>::Remove(ProtoSortedTree::Item &) Line 354	C++
 	Norm.dll!ProtoSortedTree::Remove(ProtoSortedTree::Item & item={...}) Line 1640	C++
 	Norm.dll!ProtoTimerMgr::RemoveLongTimer(ProtoTimer & theTimer={...}) Line 508	C++
 	Norm.dll!ProtoTimerMgr::OnPulseTimeout(ProtoTimer & __formal) Line 218	C++
 	[Inline Frame] Norm.dll!ProtoTimer::DoTimeout() Line 206	C++
 	Norm.dll!ProtoTimerMgr::OnSystemTimeout() Line 181	C++
 	Norm.dll!ProtoDispatcher::Dispatch() Line 2702	C++
 	Norm.dll!ProtoDispatcher::Run(bool oneShot) Line 1003	C++
 	Norm.dll!ProtoDispatcher::DoThreadStart(void * param=0x000001db52299b18) Line 1061	C++

Code：ProtoTree::Bit

        do                      
        {     
            q = x;              
            if (Bit(key, keysize, x->bit, keyEndian))
                x = x->right; // !!! x == x->right  !!!
            else                
                x = x->left;    
        } while (x != &item);

[honglei]

This infinite loop cause call to norm.dll blocked forever:

 	[Inline Frame] Norm.dll!ProtoDispatcher::Lock(_RTL_CRITICAL_SECTION &) Line 306	C++
	Norm.dll!ProtoDispatcher::SuspendThread() Line 1233	C++
 	Norm.dll!NormObjectRetain(const void * objectHandle=0x000001db5b9ce9b0) Line 2726	C++

[bebopagogo]

Hi @honglei - Do you have some test code that produces this condition? Also, the current Protolib code does not have the "x->right == x" you show here so I am a little confused about which version of NORM/Protolib code you are using?

[honglei]

I means x->right and x is the same object, ( means &(x->right) == &x), and x != &item , so dead loop occured.
The code is : https://github.com/USNavalResearchLaboratory/protolib/blob/6db401d30dd36ee4ddc777d8c626c88cdcd5a28f/src/common/protoTree.cpp#L424

Test case:
I send 10K 20-25KB files using norm, and and then remove files to another dir when they was succeed sended.
this occured after 2K~6K files sended.

[honglei]

create_test_files.py:

import os
dirPath = r'E:\PythonPrj/testFiles'

if not os.path.isdir(dirPath):
    os.mkdir(dirPath)
for i in range(1, 10000):
    with open(os.path.join(dirPath, f"文件{i:04d}.txt"), "w") as f:
        f.write(f"{i} "*5_000)

Env: Win10 x64/VC 2022 x64/Python3.10

[bebopagogo]

So you are copying the received files to another directory upon NORM_RX_OBJECT_COMPLETED notifications? Are you making any NORM API calls during that time?

[honglei]

Yes, The error occured at the NORM sender side,
At the NORM sender-side，I move the sended files( send file dir) to another directory(sended file dir) after NORM_TX_QUEUE_EMPTY or .TX_OBJECT_PURGED.
At the NORM receiver side, I move the received files(temp file name) to another directory(real file name) upon NORM_RX_OBJECT_COMPLETED notifications.

the sender side:

async def proc_sender_event(event:pynorm.event.Event):
    #print(f"{str(event)}")
    obj = event.object
    evtType:pynorm.EventType = event.type
    session:pynorm.session.Session = event.session
    linkID:int = session.conf['linkID']
    userName:str = session.conf['userName']
    
    if obj:   #and obj.handle not in (None,pynorm.NORM_OBJECT_INVALID):
        objType:ObjectType = obj.type
        info:str|None =obj.info.decode() if obj.info else None
        print(event, linkID, obj.type, info)
    else:
        print (event)
    
    if evtType == EventType.TX_OBJECT_SENT:
        if objType == ObjectType.FILE and obj.info:
            session.sendFiles.add(obj.info.decode())
            
    elif evtType == EventType.TX_QUEUE_EMPTY:
        '''
        '''
        for key, obj in session.id2obj.items():
            obj.cancel()
        session.id2obj.clear()
        #用于关闭obj的文件句柄
        del obj
        del event 
        
        #listProcOpenFiles() #用于测试是否有未关闭的文件句柄 
        for file in session.sendFiles:
            try:
                moveFile2SendedDir(file)
            except Exception as e:
                logging.warning( traceback.format_exc() )
        session.sendFiles.clear()
        files = await get_some_waitingfiles_of_link(settings.sendFileDir,linkID)
        pushFiles2session(session, waitingFiles=files)
        
    elif evtType == EventType.TX_OBJECT_PURGED:
        '''
            
        '''
        if obj.handle in session.id2obj:
            session.id2obj.pop(obj.handle)
        info = obj.info.decode()
        if info in session.sendFiles:
            session.sendFiles.remove( info )
        obj.cancel()
        del obj
        del event
        try:
            moveFile2SendedDir( info )
        except Exception as e:
            logging.warning( traceback.format_exc() )
            
        
    elif evtType == EventType.TX_FLUSH_COMPLETED:
        '''
            the NORM sender observes when it no longer has data ready for transmission has completed
        '''
        for file in session.sendFiles:
            moveFile2SendedDir(file)
        session.sendFiles.clear()        
        files = await get_some_waitingfiles_of_link(settings.sendFileDir,linkID)
        pushFiles2session(session, waitingFiles=files)
        
    elif event.type == EventType.TX_OBJECT_PURGED:
        
        # 参见 NormSetTxCacheBounds 说明 
        handle:int = obj.handle
        if objType and handle in event.session.id2obj:
            event.session.id2obj.pop(handle) 
            
        if objType == ObjectType.FILE and obj.info:
            '''
                move sended files to the sended dir
            '''
            info:str = obj.info.decode()
            obj.cancel()
            del obj
            del event
            moveFile2SendedDir(info)
            if info in session.sendFiles:
                session.sendFiles.remove(info)
            filePath = await get_one_waitingfile_of_link(settings.sendFileDir, session.conf['linkID'])
            pushFiles2session(session, waitingFiles=[filePath])

the NORM receiver side:

async def proc_recver_file_event(event:pynorm.event.Event):
    '''
         file info is {channelName}/{fileName} encoding: uft-8 
    '''
    #mqttClient:MQTTClient = mqtt_client()
    
    evtType:EventType = event.type
    session:pynorm.session.Session = event.session
    obj:pynorm.object.Object = event.object
    #linkID = session.conf['linkID']
    info:str = obj.info.decode() if obj and obj.info else None
    print (f" {evtType}  {info=}")

    if evtType == EventType.RX_OBJECT_NEW:
        now:str = datetime.datetime.now().isoformat()
        
        channelName, fileName = info.split("/")
        obj.begin = now
        obj.fileName = fileName
        obj.channelName = channelName
        session.id2obj[ info ] =  obj


    elif evtType == EventType.RX_OBJECT_COMPLETED:
        '''
        This event is posted when a receive object is completely received, including available NORM_INFO content. 
        Unless the application specifically retains the "object" handle, the indicated NormObjectHandle becomes invalid and must no longer be referenced.
        '''
        if info in session.id2obj:
            session.id2obj.pop( info )
        #obj.cancel()
        
        newFilePath:str = os.path.join(settings.recvFileDir, obj.channelName, obj.fileName )
        oldPath:str = event.object.filename.decode( locale.getpreferredencoding() )
        print (f"{oldPath=} {newFilePath}")
        try:
            #remove conflict file 
            if os.path.isfile(newFilePath):
                os.remove(newFilePath)
            newDirPath:str = os.path.join(settings.recvFileDir, obj.channelName )
            #create dest dir.
            if not os.path.isdir(newDirPath):
                os.makedirs( newDirPath )           
            #remove temp file to new dir and give the final real name. 
            os.rename(src=oldPath, dst=newFilePath)
        except Exception as e:
            logging.warning ( traceback.format_exc() )  
            
        logging.info(f"{obj.channelName, obj.fileName} completed")
        
    elif evtType == EventType.RX_OBJECT_ABORTED:
        oldPath:str = event.object.filename.decode( locale.getpreferredencoding() )
        obj.cancel()
        os.remove(oldPath) #移除 临时文件 

        logging.warning( f'{obj.channelName, obj.fileName} aborted' )    

[honglei]

Are you making any NORM API calls during that time?

Yes, user may add new send files(by calling session.fileEnqueue) while event notification process not complete.

[bebopagogo]

The TX_QUEUE_EMPTY notification should only be used as a cue for the sender application to enqueue new content for transmission.

The TX_OBJECT_PURGED notification is how the NORM lets the sender application that it is done with the associated file or memory content. NORM itself will delete the object after the notification is issued and the application does not need to (and should not) do that itself. It is an opportunity for the application dispose of (or otherwise manage) the actual file (or memory block) associated with the NormObject, but NORM itself will clean up the NormObject state.

I think the error is occurring because somehow a ProtoTree::Item is being removed from a tree twice (possibly when your code cancels/deletes the obj and then when NORM tries to also do the same) and the Item left/right pointers are not valid. I should probably "harden" the ProtoTree code a little more to avoid the loop issue here ... or at least make sure there is an ASSERT() that will fail instead of an infinite loop.

My recommendation is to remove the calls you make to:

obj.cancel()
del obj
del event

from your handling of those two notifications since the NORM code does that itself under the hood. Also you should only move the sender file during the PURGED notification.

The "norm/examples/python/normMsgr.py" illustrates the handling of TX_QUEUE_EMPTY and TX_OBJECT_PURGED notifications. Even though it is using NORM_OBJECT_DATA (memory blocks) instead of NORM_OBJECT_FILE (files), the handling of those notifications is similar. It simply is deleting the application message buffer upon purge where you want to move the sent files to another directory or something. I do plan to create a "normCast.py" example at some point that is similar to the "normCast.cpp" example. The "normMsgr.cpp" and "normMsgr.py" examples are very similar to an example of file transmission/reception.

[honglei]

1 Norm sender, 1 norm receiver, the sender side has never received a TX_OBJECT_PURGED event,
TX_QUEUE_EMPTY or TX_FLUSH_COMPLETED do received.

the events received in the NORM sender side:

GRTT_UPDATED
CC_ACTIVE
TX_RATE_CHANGED
CC_ACTIVE
GRTT_UPDATED
TX_OBJECT_SENT 1 ObjectType.FILE a中文频道1/文件0001.txt
GRTT_UPDATED
GRTT_UPDATED
TX_OBJECT_SENT 1 ObjectType.FILE a中文频道1/文件0002.txt
GRTT_UPDATED
TX_OBJECT_SENT 1 ObjectType.FILE a中文频道1/文件0003.txt
TX_OBJECT_SENT 1 ObjectType.FILE a中文频道1/文件0004.txt
TX_OBJECT_SENT 1 ObjectType.FILE a中文频道1/文件0005.txt
GRTT_UPDATED
TX_OBJECT_SENT 1 ObjectType.FILE a中文频道1/文件0006.txt
TX_OBJECT_SENT 1 ObjectType.FILE a中文频道1/文件0007.txt
TX_OBJECT_SENT 1 ObjectType.FILE a中文频道1/文件0008.txt
TX_OBJECT_SENT 1 ObjectType.FILE a中文频道1/文件0009.txt
GRTT_UPDATED
TX_OBJECT_SENT 1 ObjectType.FILE a中文频道1/文件0010.txt
TX_QUEUE_EMPTY
GRTT_UPDATED
GRTT_UPDATED
TX_FLUSH_COMPLETED

[bebopagogo]

A TX_OBJECT_PURGED notification is issued depending upon the sender tx cache parameters which consist of min count, max count, and max size. For example if min_count/max_count is 8, a purge notification for object #1 will be issued when object #9 is enqueued. For NACK-based operation, the NORM sender keeps transmit objects cached for possible response to NACKs ... the cache limit parameters determine when the oldest object is purged from the cache to make room for a newly enqueued object. (and newly enqueued objects are subject to the default timer-based flow control or if the sender app implements flow control using the ACK option)

[honglei]

So, I can manually/safely remove all send files when TX_FLUSH_COMPLETED occured?

      elif evtType == EventType.TX_FLUSH_COMPLETED:
        '''
            the NORM sender observes when it no longer has data ready for transmission has completed
        '''
        for key, obj in session.id2obj.items():
            obj.cancel()
        session.id2obj.clear()
        del obj
        
        for file in session.sendFiles:
            moveFile2SendedDir(file)
        session.sendFiles.clear()   
 
        files = await get_some_waitingfiles_of_link(settings.sendFileDir,linkID)
        pushFiles2session(session, waitingFiles=files)

[honglei]

I need to do

del obj
del event

because when TX_OBJECT_PURGED occured(the time I wanna move the file to sended file dir), the related obj file handle is still not closed by NORM:

TX_OBJECT_PURGED 1 ObjectType.FILE a中文频道1/文件0081.txt
11:44:46.996ERROR|deps.py|watch_norm_events|Traceback (most recent call last):
  File "E:\PythonPrj\crud_test\norm\deps.py", line 310, in watch_norm_events
    await proc_event(event)
  File "E:\PythonPrj\crud_test\server\proc_norm.py", line 129, in proc_sender_event
    moveFile2SendedDir(info)
  File "E:\PythonPrj\crud_test\server\proc_norm.py", line 53, in moveFile2SendedDir
    os.rename( currentFilePath, newFilePath )#移动到已发送目录
PermissionError: [WinError 32] 另一个程序正在使用此文件，进程无法访问。: '../sendFiles\\a中文频道1/文件0081.txt' -> '../sendedFiles\\a中文频道1\\文件0081.txt'

[honglei]

Even the TX_OBJECT_PURGED event process complete, the file handle is still not closed by NORM.
I try to move file after 1 seconds :

    elif event.type == EventType.TX_OBJECT_PURGED:
        # 参见 NormSetTxCacheBounds 说明 
        handle:int = obj.handle
        if objType and handle in event.session.id2obj:
            event.session.id2obj.pop(handle) 
            
        if objType == ObjectType.FILE and obj.info:
            info:str = obj.info.decode()
            #obj.cancel()
            #del obj
            #del event
            #moveFile2SendedDir(info)
            if info in session.sendFiles:
                session.sendFiles.remove(info)
            filePath = await get_one_waitingfile_of_link(settings.sendFileDir, linkID, not_in=session.sendFiles)
            pushFiles2session(session, waitingFiles=[filePath])
            loop = asyncio.get_running_loop()
            loop.call_later(1, moveFile2SendedDir, info)  

[bebopagogo]

OK ... so the event is dispatched to your application before the file is closed? I probably should update the code to close the file handle before the PURGE is issued. By the way, you could also instead use the NormFileRename() function to rename/move the file. I need to double check if that function is part of the Python binding. If not, it should be added. But it's a good idea to have the file handle closed before the PURGE notification in any case, so this is another good find by you. Thanks! I appreciate your rigor and patience.

So I just pushed a commit that adds and calls a NormFileObject::CloseFile() method just before the TX_OBJECT_PURGED notification is issued. This lets your code manipulate the file as desired. For NORM_OBJECT_DATA, the object was left "open" so the application could have the option of accessing and managing the associated memory block (e.g., freeing it if allocated by the application, etc). But for manipulating a file associated with a NORM_OBJECT_FILE, the file handle needs to be closed as you point out.