querynetbox.java

监控大型网络的软件。能够自动发现拓扑结构

			if (randomize) {
				sql += " ORDER BY random() * netboxid";
			}
			*/
			sql += " ORDER BY netboxid";
			//sql += " LIMIT 1000";
			rs = Database.query(sql);

			String prevnetboxid = null;
			Map keyFreqMap = null, keyMap = null;

			int nbHigh = netboxHigh;
			Set netboxidSet = new HashSet();
			List addToRunQList = new ArrayList();

			while (rs.next()) {
				String netboxid = rs.getString("netboxid");

				try {
					if (!netboxid.equals(prevnetboxid)) {
						keyFreqMap = new HashMap();
						keyMap = new HashMap();
						addDefaultOids(keyFreqMap, keyMap);

						String typeid = rs.getString("typeid");
						if (typeid == null) typeid = Type.UNKNOWN_TYPEID;
						Type t = (Type)typeidMap.get(typeid);
						if (t == null) {
							Log.d("UPDATE_NETBOXES", "Skipping netbox " + rs.getString("sysname") +
								  " because type is null (probably the type doesn't have any OIDs)");
							skipcnt++;
							continue;
						}
						NetboxImpl nb;

						boolean newNetbox = false;
						if ( (nb=(NetboxImpl)nbMap.get(netboxid)) == null) {
							nbMap.put(netboxid, nb = new NetboxImpl(++nbHigh, t, keyFreqMap, keyMap));
							newNetbox = true;
						} else {
							long oldNextRun = nb.getNextRun();
							nb.setType(t);
							if (oldNextRun != nb.getNextRun()) {
								// We need to remove the netbox from the runq and re-insert it
								removeFromRunQ(nb, new Long(oldNextRun));
								newNetbox = true;
							}
						}

						nb.setDeviceid(rs.getInt("deviceid"));
						nb.setNetboxid(netboxid);
						nb.setIp(rs.getString("ip"));
						nb.setCommunityRo(rs.getString("ro"));
						nb.setSysname(rs.getString("sysname"));
						nb.setCat(rs.getString("catid"));
						int numInStack = 1;
						if (numInStackMap.containsKey(netboxid)) numInStack = Integer.parseInt((String)numInStackMap.get(netboxid));
						nb.setNumInStack(numInStack);
						nb.setSnmpVersion(rs.getInt("snmp_version"));
						//nb.setSnmpagent(rs.getString("snmpagent"));

						///////////////////////////////////////////
						if (newNetbox) {
							newcnt++;
							addToRunQList.add(nb);
							if (!rs.getBoolean("uptodate")) {
								synchronized (deviceNetboxCache) {
									deviceNetboxCache.remove(nb.getNetboxidS());
								}
							}
						}
						netboxidSet.add(new Integer(nb.getNetboxid()));
						nb.setUptodate(rs.getBoolean("uptodate"));
						/*
						netboxidSet.add(new Integer(nb.getNetboxid()));
						//System.err.println("Sysname: " + rs.getString("sysname") + ", " + rs.getBoolean("uptodate") + ", " + oidNetboxSet);
						if (!rs.getBoolean("uptodate") && oidNetboxSet.add(netboxid)) {
							synchronized (deviceNetboxCache) {
								deviceNetboxCache.remove(nb.getNetboxidS());
							}
							addToOidQList.add(nb);
							System.err.println("Added to oidq: " + nb);
							oidnew++;
						}
						*/
						///////////////////////////////////////////

					}

					String oidkey = rs.getString("oidkey");
					if (oidkey != null) {
						Snmpoid snmpoid = (Snmpoid)oidkeyMap.get(oidkey);

						boolean oidfreq = (rs.getString("oidfreq") != null && rs.getString("oidfreq").length() > 0);
						int freq = oidfreq ? rs.getInt("oidfreq") : rs.getInt("typefreq");
						if (freq <= 0) {
							Log.w("UPDATE_NETBOXES", "No frequency specified for netbox " + netboxid + ", oid: " + oidkey + ", skipping.");
							continue;
						}
						keyFreqMap.put(oidkey, new Integer(freq));
						keyMap.put(oidkey, snmpoid);
					} else {
						Log.d("UPDATE_NETBOXES", "No OIDs found for netbox: " + rs.getString("sysname"));
					}
				} finally {
					prevnetboxid = netboxid;
				}
			}
				
			for (Iterator it = addToRunQList.iterator(); it.hasNext();) {
				NetboxImpl nb = (NetboxImpl)it.next();
				nb.updateNextRun();
				initialQ.add(nb);
				//addToRunQ(nb);
			}

			netboxCnt = netboxidSet.size();
			netboxHigh = nbHigh;

			// Remove netboxes no longer present
			for (Iterator it = nbMap.values().iterator(); it.hasNext();) {
				NetboxImpl nb = (NetboxImpl)it.next();
				if (!netboxidSet.contains(new Integer(nb.getNetboxid()))) {
					nb.remove();
					it.remove();
					delcnt++;
				}
			}

			// Re-create list
			if (newcnt > 0 || delcnt > 0) {
				List nbL = new ArrayList(nbMap.values());
				nbList = nbL;
			}

		} catch (SQLException e) {
			Log.e("UPDATE_NETBOXES", "SQLException: " + e);			
			e.printStackTrace(System.err);
		}

		String msg = "Num netboxes: " + netboxCnt + " (" + netboxHigh + " high, " + newcnt + " new, " + delcnt + " removed, " + skipcnt + " skipped, " + nbRunQSize() + " runq,";
		if (initialQ.size() > 0) msg += " " + initialQ.size() + " initialQ,";
		msg +=  " " + threadCnt + " active)";
		Log.i("UPDATE_NETBOXES", msg);
		//Log.i("UPDATE_NETBOXES", "Num netboxes: " + netboxCnt + " (" + netboxHigh + " high, " + newcnt + " new, " + delcnt + " removed, " + skipcnt + " skipped, " + nbRunQSize() + " runq, " + threadCnt + " active)");

		// Check the run queue in case we have any new netboxes to check
		if (newcnt > 0) {
			scheduleCheckRunQ(0);
		}

	}

	private static void addToRunQ(NetboxImpl nb) {
		addToRunQ(nb, false);
	}

	private static void addToRunQFront(NetboxImpl nb) {
		addToRunQ(nb, true);
	}

	private static void addToRunQ(NetboxImpl nb, boolean front) {
		Long nextRun = new Long(nb.getNextRun());
		synchronized (nbRunQ) {
 			LinkedList l;
			if ( (l = (LinkedList)nbRunQ.get(nextRun)) == null) nbRunQ.put(nextRun, l = new LinkedList());
			if (front) {
				l.addFirst(nb);
			} else {
				l.add(nb);
			}
			netboxidRunQMap.put(nb.getNetboxidS(), nextRun);
		}
	}

	private static void removeFromRunQ(NetboxImpl nb, Long nextRun) {
		synchronized (nbRunQ) {
 			LinkedList l;
			if ( (l = (LinkedList)nbRunQ.get(nextRun)) == null) {
				// Check if the netbox is still in the initialQ
				for (Iterator it = initialQ.iterator(); it.hasNext();) {
					if (nb.getNum() == ((NetboxImpl)it.next()).getNum()) {
						it.remove();
						break;
					}
				}
				return;
			}
			for (Iterator it = l.iterator(); it.hasNext();) {
				if (nb.getNum() == ((NetboxImpl)it.next()).getNum()) {
					it.remove();
					netboxidRunQMap.remove(nb.getNetboxidS());
					break;
				}
			}
			if (l.isEmpty()) nbRunQ.remove(nextRun);
		}
	}

	private static Object removeRunQHead() {
		Object o;
		while ((o = removeRunQHeadNoCheck()) instanceof NetboxImpl) {
			NetboxImpl nb = (NetboxImpl)o;
			if (nb.isRemoved()) continue;
			return nb;
		}
		return o;
	}

	private static int nbRunQSize() {
		synchronized (nbRunQ) {
			return nbRunQ.size();
		}
	}

	private static Object removeRunQHeadNoCheck() {
		synchronized (nbRunQ) {
			boolean isInitial = !initialQ.isEmpty();
			boolean rand = Math.random() >= 0.5f; // 50% of picking initial if runQ is not empty
			if (extraThreadCnt > 0) rand = false; // Don't pick from initial queue when doing immediate netbox runs
			if (nbRunQ.isEmpty() || (isInitial && rand)) {
				if (!initialQ.isEmpty()) return initialQ.remove(initialQ.size()-1);
				return new Long(Long.MAX_VALUE / 2); // Infinity...
			}
			Long nextRun = (Long)nbRunQ.firstKey();
			if (nextRun.longValue() > System.currentTimeMillis()) {
				// Head of queue is not yet ready to be run
				if (!initialQ.isEmpty()) return initialQ.remove(initialQ.size()-1);
				return new Long(nextRun.longValue() - System.currentTimeMillis());
			}

			LinkedList l = (LinkedList)nbRunQ.get(nextRun);
			NetboxImpl nb  = (NetboxImpl)l.removeFirst();
			netboxidRunQMap.remove(nb.getNetboxidS());
			if (l.isEmpty()) nbRunQ.remove(nextRun);
			return nb;
		}
	}

	private static NetboxImpl removeFromRunQ(String netboxid) {
		synchronized (nbRunQ) {
 			LinkedList l;
			Long nextRun = (Long)netboxidRunQMap.get(netboxid);
			if (nextRun == null) {
				// Check if the netbox is still in the initialQ
				for (Iterator it = initialQ.iterator(); it.hasNext();) {
					NetboxImpl nb = (NetboxImpl)it.next();
					if (nb.getNetboxidS().equals(netboxid)) {
						it.remove();
						return nb;
					}
				}
				Log.d("QUERY_NETBOX", "REMOVE_FROM_RUNQ", "nextRun not found for netboxid " + netboxid);
				return null;
			}
			if ( (l = (LinkedList)nbRunQ.get(nextRun)) == null) return null;
			for (Iterator it = l.iterator(); it.hasNext();) {
				NetboxImpl nb = (NetboxImpl)it.next();
				if (nb.getNetboxidS().equals(netboxid)) {
					it.remove();
					if (l.isEmpty()) nbRunQ.remove(nextRun);
					netboxidRunQMap.remove(nb.getNetboxidS());
					return nb;
				}
			}
		}
		return null;
	}

	// Run the nexbox immediately
	private static boolean runNetbox(String netboxid, String maxage, String source, String subid) {
		scheduleImmediatelyMap.put(netboxid, new String[] { source, subid });
		NetboxImpl nb = removeFromRunQ(netboxid);
		Log.d("QUERY_NETBOX", "RUN_NETBOX", "Immediate run for netbox("+netboxid+"): " + nb);
		if (nb == null) return false;
		nb.scheduleImmediately(); // Schedule immediately
		addToRunQFront(nb);
		synchronized (idleThreadLock) {
			// If there are no idle threads, make one
			extraThreadCnt++;
		}
		scheduleCheckRunQ(0);
		return true;
	}

	private static String requestThread(boolean allowExtra) {
		synchronized (idleThreadLock) {
			int max = allowExtra ? maxThreadCnt+extraThreadCnt : maxThreadCnt;
			if (allowExtra && extraThreadCnt > 0) extraThreadCnt--;
			if (threadCnt < max) {
				int tid = getFreeTid();
				threadCnt++;
				//System.err.println("New thread, cnt="+(threadCnt+1)+" max="+max);
				return format(tid, String.valueOf(max-1).length());
			}
			return null;
		}
	}

	private static void threadIdle(String tid) {
		synchronized (idleThreadLock) {
			returnTid(Integer.parseInt(tid));
			//System.err.println("Del thread, cnt="+(threadCnt-1));
			threadCnt--;
		}
		scheduleCheckRunQ(0);
	}

	private static int getFreeTid() {
		if (threadIdList.isEmpty()) {
			threadIdList.add(new Integer(curMaxTid++));
		}
		return ((Integer)threadIdList.removeFirst()).intValue();
	}

	private static void returnTid(int tid) {
		threadIdList.add(new Integer(tid));
	}

	// Constructor
	public QueryNetbox(String tid, NetboxImpl initialNb)
	{
		this.tid = tid;
		this.setName("QueryNetbox-" + tid);
		this.nb = initialNb;
	}

	public QueryNetbox(String tid, Object oidUpdObj) {
		this.tid = tid;
		this.setName("QueryNetbox-" + tid);
		this.oidUpdObj = oidUpdObj;
	}

	public void run()
	{
		Log.setDefaultSubsystem("QUERY_NETBOX_T"+tid);
		Log.setThreadId(tid);

		Log.d("RUN", "Thread " + tid + " starting work on ("+nb+"|"+oidUpdObj+")");

		long beginTime = System.currentTimeMillis();
		SimpleSnmp sSnmp = null;

		try {
			while (true) {

				// Check if we were assigned an oid object and not a netbox
				if (oidUpdObj == null && !nb.getUptodate()) oidUpdObj = nb;

				if (oidUpdObj != null) {
					OidTester oidTester = new OidTester();
					if (oidUpdObj instanceof NetboxImpl) {
						nb = (NetboxImpl)oidUpdObj;
						sSnmp = SimpleSnmp.simpleSnmpFactory(nb.getTypeT().getVendor(), nb.getType());
						sSnmp.setHost(nb.getIp());
						sSnmp.setCs_ro(nb.getCommunityRo());
						sSnmp.checkSnmpVersion();
						oidTester.oidTest(nb, oidkeyMap.values().iterator(), sSnmp );
						//nb.scheduleImmediately();
					} else if (oidUpdObj instanceof Snmpoid) {
						Snmpoid oid = (Snmpoid)oidUpdObj;
						oidTester.oidTest(oid, nbList.iterator() );
						oidKeySet.remove(oid.getOidkey());
						synchronized (oidQ) {
							if (oidQ.isEmpty()) {
								Log.d("RUN", "oidQ empty, scheduling types/netboxes update");