multi Threading queue with netmiko

Viewed 168

im creating this script to process multiple équipements with netmiko , by using threading queue the problem is the execution of the script dont finished and the last message process completed , not showing, only when the equipement queue ( csg_queu) size is > number of threads . example here i put 6 equipements in the csg_list , so the number of threads will be 5 result the script running infinitely, if i put 5 equipement or less its working, so can any one help me to find the issue plz

# coding: utf-8
from netmiko import ConnectHandler, NetmikoTimeoutException,redispatch
from paramiko import AuthenticationException
import time
import queue
from threading import Thread, Lock, currentThread


def lecture_csg_list():
    with open("csg_list.txt", "r") as csg_list:
        listcsg = csg_list.readlines()
        listcsg = list(map(lambda x: x.strip(), listcsg))
    return listcsg


def mt(mt_function, q):
    if q.qsize() > 5:
        num_threads = 5
    else:
        num_threads = q.qsize()
    for i in range(0, num_threads):
        thread_name = f'Thread-{i}'
        worker = Thread(name=thread_name, target=mt_function, args=(q,))
        worker.setDaemon(True)
        worker.start()  
        
    q.join()


def csg_connect(csg):
    error = "none"
    csg_vendor = "none"
    counter = 0
    bt1shz07 = {'device_type': 'terminal_server', 'ip': 'XXXXXXXXXXXX', 'username': user, 'password': password, }
    csg_net_connect = ConnectHandler(**bt1shz07)
    csg_net_connect.write_channel('ssh ' + csg + "\n")
    if "CSG1" in csg:
        csg_vendor = "nokia"
        while True:
            output = csg_net_connect.read_channel()
            if "password" in output:
                csg_net_connect.write_channel(csg_net_connect.password + "\n")
                time.sleep(0.5)
                break
            if "Received disconnect from" in output:
                error = "Unreachable"
                return error, csg_vendor
            if counter == 100000000:
                error = "Connecting time out"
                return error, csg_vendor
            counter += 1

        redispatch(csg_net_connect, device_type='nokia_sros')
        time.sleep(1)
        csg_net_connect.disconnect()
    else:
        csg_vendor = "cisco"
        while True:
            output = csg_net_connect.read_channel()
            if "Password" in output:
                csg_net_connect.write_channel(csg_net_connect.password + "\n")
                time.sleep(0.5)
                break
            if "Received disconnect from" in output:
                error = "Unreachable"
                return error, csg_vendor
            if counter == 100000000:
                error = "Connecting time out"
                return error, csg_vendor
            counter += 1

        redispatch(csg_net_connect, device_type='cisco_ios')
        time.sleep(1)
        csg_net_connect.disconnect()
        
    return error, csg_vendor


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

def check_csg_state(q):
    global nbr
    while True:
        if q.empty():
            print(currentThread().getName()," : terminated")
            q.task_done()
            return
        thread_name = currentThread().getName()
        csg_host = q.get()

        error, csg_vendor = csg_connect(csg_host)

        if error == "none":
            with PRINT_LOCK:
                print(str(nbr)+"-"+csg_host+":"+csg_vendor+"   OK    :",currentThread().getName())
                nbr+=1
                
        else:

            with PRINT_LOCK:
                print(str(nbr)+"-"+csg_host + ":" + csg_vendor + "   NOK     :",currentThread().getName())
                nbr+=1

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





if __name__ == '__main__':

    PRINT_LOCK = Lock()
    nbr = 1
    user = "XXXXXXXXXXXXXXXX"
    password = "XXXXXXXXXXXXXXXXXX"
    csg_list = lecture_csg_list()
    csg_queue = queue.Queue()
    for csg in csg_list:
         csg_queue.put(csg)
    mt(mt_function=check_csg_state, q=csg_queue)
    print("Process Completed ...")
    
1 Answers

I found the issue , it was the position of q.task_done() in the function check_csg_state() so the correction

def check_csg_state(q):
    global nbr
    while True:
        If q.empty()
            return
        thread_name = currentThread().getName()
        csg_host = q.get()

        error, csg_vendor = csg_connect(csg_host)

        if error == "none":
            with PRINT_LOCK:
                print(str(nbr)+"-"+csg_host+":"+csg_vendor+"   OK    :",currentThread().getName())
                nbr+=1
                q.task_done()
        else:

            with PRINT_LOCK:
                print(str(nbr)+"-"+csg_host + ":" + csg_vendor + "   NOK     :",currentThread().getName())
                nbr+=1
                q.task_done()
Related