output: #fixed code
                                @Override
    public void run()
    {
      while (this.running)
      {
        //input
        while (!this.inputs.isEmpty())
        {
          DatagramPacket dp = this.inputs.remove();
          KcpOnUdp ku = this.kcps.get(dp.sender());
          if (ku == null)
          {
            ku = new KcpOnUdp(this.out, dp.sender(), this.listerner);//初始化
            ku.noDelay(nodelay, interval, resend, nc);
            ku.wndSize(sndwnd, rcvwnd);
            ku.setMtu(mtu);
            ku.setTimeout(timeout);
            this.kcps.put(dp.sender(), ku);
          }
          ku.input(dp.content());
        }
        //update
        KcpOnUdp temp = null;
        for (KcpOnUdp ku : this.kcps.values())
        {
          ku.update();
          if (ku.isClosed())
          {
            temp = ku;
          }
        }
        if (temp != null)//删掉过时的kcp
        {
          this.kcps.remove((InetSocketAddress) temp.getKcp().getUser());
        }
        try
        {
          Thread.sleep(this.interval);
        } catch (InterruptedException ex)
        {
        }
      }
    }
input: #vulnerable code 
                                @Override
    public void run()
    {
      while (this.running)
      {
        //input
        while (!this.inputs.isEmpty())
        {
          DatagramPacket dp = this.inputs.remove();
          KcpOnUdp ku = this.kcps.get(dp.sender());
          if (ku == null)
          {
            ku = new KcpOnUdp(this.out, dp.sender(), this.listerner);//初始化
            ku.noDelay(nodelay, interval, resend, nc);
            ku.wndSize(sndwnd, rcvwnd);
            ku.setMtu(mtu);
            ku.setTimeout(timeout);
            this.kcps.put(dp.sender(), ku);
            pqueue.add(ku);
          }
          ku.input(dp.content());
        }
        
        //选出第一个kcp更新状态
        KcpOnUdp first = pqueue.poll();
        if(first != null && first.getTimeout() < System.currentTimeMillis()) {
            first.update();
            if(!first.isClosed()) {
                pqueue.add(first);
            } else {
                this.kcps.remove((InetSocketAddress) first.getKcp().getUser());
            }
        }
       
        //每30s，更新一遍所有的kcp状态
        if(System.currentTimeMillis()%(1000*30) == 0) {
          //update
          KcpOnUdp temp = null;
          for (KcpOnUdp ku : this.kcps.values())
          {
            ku.update();
            if (ku.isClosed()) {//删掉过时的kcp
              this.kcps.remove((InetSocketAddress) temp.getKcp().getUser());
              pqueue.remove(ku);
            }
          }
        }
        
      //等待 
      try {  
          synchronized(wakeup){  
            wakeup.wait(5*60*1000);
          }
      } catch (InterruptedException ex) {
          Logger.getLogger(KcpThread.class.getName()).log(Level.SEVERE, null, ex);
      }
         
      }
    } 
                              #location 4 
                              #vulnerability type THREAD_SAFETY_VIOLATION